Added a 2 step compression process. First compress incoming traffic, then send...

Added a 2 step compression process. First compress incoming traffic, then send "go ahead" stanza to client and finally compress outgoing traffic. JM-1059

git-svn-id: http://svn.igniterealtime.org/svn/repos/openfire/trunk@8352 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/openfire/Connection.java

@@ -258,8 +258,16 @@ public interface Connection {
     void startTLS(boolean clientMode, String remoteServer) throws Exception;
 
     /**
-     * Start using compression for this connection. Compression will only be available after TLS
-     * has been negotiated. This means that a connection can never be using compression before
+     * Adds the compression filter to the connection but only filter incoming traffic. Do not filter
+     * outgoing traffic since we still need to send an uncompressed stanza to the client indicating
+     * that he can start compressing the traffic. After we sent the uncompresses stanza we can
+     * start compression outgoing traffic as well.
+     */
+    void addCompression();
+
+    /**
+     * Start compressing outgoing traffic for this connection. Compression will only be available after
+     * TLS has been negotiated. This means that a connection can never be using compression before
      * TLS. However, it is possible to use compression without TLS.
      */
     void startCompression();

src/java/org/jivesoftware/openfire/net/SocketConnection.java

@@ -13,9 +13,6 @@ package org.jivesoftware.openfire.net;
 
 import com.jcraft.jzlib.JZlib;
 import com.jcraft.jzlib.ZOutputStream;
-import org.jivesoftware.util.JiveGlobals;
-import org.jivesoftware.util.LocaleUtils;
-import org.jivesoftware.util.Log;
 import org.jivesoftware.openfire.Connection;
 import org.jivesoftware.openfire.ConnectionCloseListener;
 import org.jivesoftware.openfire.PacketDeliverer;
@@ -23,6 +20,9 @@ import org.jivesoftware.openfire.PacketException;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.session.IncomingServerSession;
 import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.util.JiveGlobals;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
 import org.xmpp.packet.Packet;
 
 import javax.net.ssl.SSLSession;
@@ -168,6 +168,10 @@ public class SocketConnection implements Connection {
         }
     }
 
+    public void addCompression() {
+        // WARNING: We do not support adding compression for incoming traffic but not for outgoing traffic.
+    }
+
     public void startCompression() {
         compressed = true;
 

src/java/org/jivesoftware/openfire/net/SocketReadingMode.java

@@ -13,9 +13,9 @@ package org.jivesoftware.openfire.net;
 
 import org.dom4j.DocumentException;
 import org.dom4j.Element;
-import org.jivesoftware.util.Log;
 import org.jivesoftware.openfire.Connection;
 import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.util.Log;
 import org.xmlpull.v1.XmlPullParserException;
 import org.xmpp.packet.StreamError;
 
@@ -196,10 +196,13 @@ abstract class SocketReadingMode {
             return false;
         }
         else {
+            // Start using compression for incoming traffic
+            socketReader.connection.addCompression();
+
             // Indicate client that he can proceed and compress the socket
             socketReader.connection.deliverRawText("<compressed xmlns='http://jabber.org/protocol/compress'/>");
 
-            // Start using compression
+            // Start using compression for outgoing traffic
             socketReader.connection.startCompression();
             return true;
         }

src/java/org/jivesoftware/openfire/net/StanzaHandler.java

@@ -12,13 +12,13 @@ package org.jivesoftware.openfire.net;
 
 import org.dom4j.Element;
 import org.dom4j.io.XMPPPacketReader;
-import org.jivesoftware.util.LocaleUtils;
-import org.jivesoftware.util.Log;
-import org.jivesoftware.util.StringUtils;
 import org.jivesoftware.openfire.Connection;
 import org.jivesoftware.openfire.PacketRouter;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.util.StringUtils;
 import org.xmlpull.v1.XmlPullParser;
 import org.xmlpull.v1.XmlPullParserException;
 import org.xmpp.packet.*;
@@ -450,10 +450,13 @@ public abstract class StanzaHandler {
             connection.deliverRawText(error);
             return false;
         } else {
+            // Start using compression for incoming traffic
+            connection.addCompression();
+
             // Indicate client that he can proceed and compress the socket
             connection.deliverRawText("<compressed xmlns='http://jabber.org/protocol/compress'/>");
 
-            // Start using compression
+            // Start using compression for outgoing traffic
             connection.startCompression();
             return true;
         }

src/java/org/jivesoftware/openfire/net/VirtualConnection.java

@@ -11,12 +11,12 @@
 
 package org.jivesoftware.openfire.net;
 
-import org.jivesoftware.util.LocaleUtils;
-import org.jivesoftware.util.Log;
 import org.jivesoftware.openfire.Connection;
 import org.jivesoftware.openfire.ConnectionCloseListener;
 import org.jivesoftware.openfire.PacketDeliverer;
 import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -111,6 +111,10 @@ public abstract class VirtualConnection implements Connection {
         //Ignore
     }
 
+    public void addCompression() {
+        //Ignore
+    }
+
     public void startCompression() {
         //Ignore
     }

src/java/org/jivesoftware/openfire/nio/NIOConnection.java

@@ -314,13 +314,18 @@ public class NIOConnection implements Connection {
         }
     }
 
-    public void startCompression() {
+    public void addCompression() {
         IoFilterChain chain = ioSession.getFilterChain();
         String baseFilter = "org.apache.mina.common.ExecutorThreadModel";
         if (chain.contains("tls")) {
             baseFilter = "tls";
         }
-        chain.addAfter(baseFilter, "compression", new CompressionFilter(CompressionFilter.COMPRESSION_MAX));
+        chain.addAfter(baseFilter, "compression", new CompressionFilter(true, false, CompressionFilter.COMPRESSION_MAX));
+    }
+
+    public void startCompression() {
+        CompressionFilter ioFilter = (CompressionFilter) ioSession.getFilterChain().get("compression");
+        ioFilter.setCompressOutbound(true);
     }
 
     public boolean isFlashClient() {

src/java/org/jivesoftware/openfire/session/OutgoingServerSession.java

@@ -16,9 +16,6 @@ import com.jcraft.jzlib.ZInputStream;
 import org.dom4j.DocumentException;
 import org.dom4j.Element;
 import org.dom4j.io.XMPPPacketReader;
-import org.jivesoftware.util.JiveGlobals;
-import org.jivesoftware.util.Log;
-import org.jivesoftware.util.StringUtils;
 import org.jivesoftware.openfire.*;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.net.DNSUtil;
@@ -29,6 +26,9 @@ import org.jivesoftware.openfire.server.RemoteServerConfiguration;
 import org.jivesoftware.openfire.server.RemoteServerManager;
 import org.jivesoftware.openfire.server.ServerDialback;
 import org.jivesoftware.openfire.spi.BasicStreamIDFactory;
+import org.jivesoftware.util.JiveGlobals;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.util.StringUtils;
 import org.xmlpull.v1.XmlPullParser;
 import org.xmlpull.v1.XmlPullParserException;
 import org.xmpp.packet.*;
@@ -403,6 +403,7 @@ public class OutgoingServerSession extends Session {
                             Element answer = reader.parseDocument().getRootElement();
                             if ("compressed".equals(answer.getName())) {
                                 // Server confirmed that we can use zlib compression
+                                connection.addCompression();
                                 connection.startCompression();
                                 Log.debug("OS - Stream compression was successful with " + hostname);
                                 // Stream compression was successful so initiate a new stream
@@ -569,10 +570,7 @@ public class OutgoingServerSession extends Session {
                 reply.setFrom(packet.getTo());
                 reply.setChildElement(((IQ) packet).getChildElement().createCopy());
                 reply.setError(PacketError.Condition.remote_server_not_found);
-                ChannelHandler route = routingTable.getRoute(reply.getTo());
-                if (route != null) {
-                    route.process(reply);
-                }
+                routingTable.routePacket(reply.getTo(), reply);
             }
             else if (packet instanceof Presence) {
                 Presence reply = new Presence();
@@ -580,10 +578,7 @@ public class OutgoingServerSession extends Session {
                 reply.setTo(packet.getFrom());
                 reply.setFrom(packet.getTo());
                 reply.setError(PacketError.Condition.remote_server_not_found);
-                ChannelHandler route = routingTable.getRoute(reply.getTo());
-                if (route != null) {
-                    route.process(reply);
-                }
+                routingTable.routePacket(reply.getTo(), reply);
             }
             else if (packet instanceof Message) {
                 Message reply = new Message();
@@ -593,10 +588,7 @@ public class OutgoingServerSession extends Session {
                 reply.setType(((Message)packet).getType());
                 reply.setThread(((Message)packet).getThread());
                 reply.setError(PacketError.Condition.remote_server_not_found);
-                ChannelHandler route = routingTable.getRoute(reply.getTo());
-                if (route != null) {
-                    route.process(reply);
-                }
+                routingTable.routePacket(reply.getTo(), reply);
             }
         }
         catch (UnauthorizedException e) {
@@ -664,7 +656,7 @@ public class OutgoingServerSession extends Session {
             // was already registered nothing happens
             sessionManager.registerOutgoingServerSession(hostname, this);
             // Add a new route for this new session
-            XMPPServer.getInstance().getRoutingTable().addRoute(new JID(hostname), this);
+            XMPPServer.getInstance().getRoutingTable().addServerRoute(new JID(hostname), this);
         }
     }