Fixed stream compression. Tested only with Pandion (Exodus pending). JM-493

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

build/lib/versions.txt

@@ -18,6 +18,7 @@ jdic.jar             | 0.9.1 (for windows only)
 jstl.jar             | Jakarta standard taglib 1.1.2
 jmdns.jar            | 1.0 RC1
 jtds.jar             | 1.2
+jzlib.jar            | 1.0.7
 mysql.jar            | 3.1.12
 pack200task.jar      | August 5, 2004
 postgres.jar         | 8.1-404 JDBC 3

src/java/org/jivesoftware/wildfire/net/MXParser.java

@@ -15,6 +15,7 @@ import org.xmlpull.v1.XmlPullParserException;
 import org.xmlpull.v1.XmlPullParser;
 
 import java.io.IOException;
+import java.io.Reader;
 
 /**
  * MXParser that returns an IGNORABLE_WHITESPACE event when a whitespace character or a
@@ -323,4 +324,12 @@ public class MXParser extends org.xmlpull.mxp1.MXParser {
             }
         }
     }
+
+    protected void resetInput() {
+        Reader oldReader = reader;
+        String oldEncoding = inputEncoding;
+        reset();
+        reader = oldReader;
+        inputEncoding = oldEncoding;
+    }
 }

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

@@ -11,14 +11,16 @@
 
 package org.jivesoftware.wildfire.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.wildfire.*;
-import org.jivesoftware.wildfire.server.IncomingServerSession;
 import org.jivesoftware.wildfire.auth.UnauthorizedException;
 import org.jivesoftware.wildfire.interceptor.InterceptorManager;
 import org.jivesoftware.wildfire.interceptor.PacketRejectedException;
-import org.jivesoftware.util.JiveGlobals;
-import org.jivesoftware.util.LocaleUtils;
-import org.jivesoftware.util.Log;
+import org.jivesoftware.wildfire.server.IncomingServerSession;
 import org.xmpp.packet.Packet;
 
 import javax.net.ssl.SSLSession;
@@ -33,7 +35,6 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.zip.ZipOutputStream;
 
 /**
  * An object to track the state of a XMPP client-server session.
@@ -74,7 +75,7 @@ public class SocketConnection implements Connection {
 
     private Session session;
     private boolean secure;
-	private boolean compressed;
+    private boolean compressed;
     private org.jivesoftware.util.XMLWriter xmlSerializer;
     private boolean flashClient = false;
     private int majorVersion = 1;
@@ -89,10 +90,10 @@ public class SocketConnection implements Connection {
      */
     private TLSPolicy tlsPolicy = TLSPolicy.optional;
 
-	/**
-	 * Compression policy currently in use for this connection.
-	 */
-	private CompressionPolicy compressionPolicy = CompressionPolicy.disabled;
+    /**
+     * Compression policy currently in use for this connection.
+     */
+    private CompressionPolicy compressionPolicy = CompressionPolicy.disabled;
 
     public static Collection<SocketConnection> getInstances() {
         return instances.keySet();
@@ -163,12 +164,16 @@ public class SocketConnection implements Connection {
         compressed = true;
 
         if (tlsStreamHandler == null) {
-            writer = new BufferedWriter(
-                    new OutputStreamWriter(new ZipOutputStream(socket.getOutputStream()), CHARSET));
+            ZOutputStream out = new ZOutputStream(socket.getOutputStream(), JZlib.Z_BEST_COMPRESSION);
+            out.setFlushMode(JZlib.Z_PARTIAL_FLUSH);
+            writer = new BufferedWriter(new OutputStreamWriter(out, CHARSET));
+            xmlSerializer = new XMLSocketWriter(writer, this);
         }
         else {
-            writer = new BufferedWriter(new OutputStreamWriter(
-                    new ZipOutputStream(tlsStreamHandler.getOutputStream()), CHARSET));
+            ZOutputStream out = new ZOutputStream(tlsStreamHandler.getOutputStream(), JZlib.Z_BEST_COMPRESSION);
+            out.setFlushMode(JZlib.Z_PARTIAL_FLUSH);
+            writer = new BufferedWriter(new OutputStreamWriter(out, CHARSET));
+            xmlSerializer = new XMLSocketWriter(writer, this);
         }
     }
 

src/java/org/jivesoftware/wildfire/net/SocketReader.java

@@ -11,17 +11,19 @@
 
 package org.jivesoftware.wildfire.net;
 
+import com.jcraft.jzlib.JZlib;
+import com.jcraft.jzlib.ZInputStream;
 import org.dom4j.DocumentException;
 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.wildfire.*;
 import org.jivesoftware.wildfire.auth.UnauthorizedException;
 import org.jivesoftware.wildfire.interceptor.InterceptorManager;
 import org.jivesoftware.wildfire.interceptor.PacketRejectedException;
 import org.jivesoftware.wildfire.server.OutgoingSessionPromise;
-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.xmlpull.v1.XmlPullParserFactory;
@@ -33,7 +35,6 @@ import java.io.InputStreamReader;
 import java.net.Socket;
 import java.net.SocketException;
 import java.nio.channels.AsynchronousCloseException;
-import java.util.zip.ZipInputStream;
 
 /**
  * A SocketReader creates the appropriate {@link Session} based on the defined namespace in the
@@ -276,8 +277,8 @@ public abstract class SocketReader implements Runnable {
                     session = null;
                 }
             }
-			else if ("compress".equals(tag)) 
-			{
+            else if ("compress".equals(tag))
+            {
                 // Client is trying to initiate compression
                 if (compressClient(doc)) {
                     // Compression was successful so open a new stream and offer
@@ -288,9 +289,9 @@ public abstract class SocketReader implements Runnable {
                     open = false;
                     session = null;
                 }
-			}
-            else 
-			{
+            }
+            else
+            {
                 if (!processUnknowPacket(doc)) {
                     Log.warn(LocaleUtils.getLocalizedString("admin.error.packet.tag") +
                             doc.asXML());
@@ -358,6 +359,9 @@ public abstract class SocketReader implements Runnable {
             return false;
         }
         else {
+            // Indicate client that he can proceed and compress the socket
+            connection.deliverRawText("<compressed xmlns='http://jabber.org/protocol/compress'/>");
+
             // Start using compression
             connection.startCompression();
             return true;
@@ -717,23 +721,15 @@ public abstract class SocketReader implements Runnable {
      * to servers or external components)
      */
     private void saslSuccessful() throws XmlPullParserException, IOException {
-		XmlPullParser xpp = reader.getXPPParser();
-		// Reset the parser since a new stream header has been sent from the client
-		if (connection.getTLSStreamHandler() == null) 
-		{
-			xpp.setInput(new InputStreamReader(socket.getInputStream(), CHARSET));
-		}
-		else 
-		{
-			xpp.setInput(new InputStreamReader(connection.getTLSStreamHandler().getInputStream(),
-				CHARSET));
-		}
-
-		// Skip the opening stream sent by the client
-		for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;) 
-		{
-			eventType = xpp.next();
-		}
+        MXParser xpp = reader.getXPPParser();
+        // Reset the parser since a new stream header has been sent from the client
+        xpp.resetInput();
+
+        // Skip the opening stream sent by the client
+        for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;)
+        {
+            eventType = xpp.next();
+        }
 
         StringBuilder sb = new StringBuilder(420);
         sb.append(geStreamHeader());
@@ -749,51 +745,52 @@ public abstract class SocketReader implements Runnable {
         connection.deliverRawText(sb.toString());
     }
 
-	/**
-	 * After compression was successful we should open a new stream and offer
-	 * new stream features such as resource binding and session establishment. Notice that
-	 * resource binding and session establishment should only be offered to clients (i.e. not
-	 * to servers or external components)
-	 */
-	private void compressionSuccessful() throws XmlPullParserException, IOException
-	{
-		connection.deliverRawText("<compressed xmlns='http://jabber.org/protocol/compress'/>");
-
+    /**
+     * After compression was successful we should open a new stream and offer
+     * new stream features such as resource binding and session establishment. Notice that
+     * resource binding and session establishment should only be offered to clients (i.e. not
+     * to servers or external components)
+     */
+    private void compressionSuccessful() throws XmlPullParserException, IOException
+    {
         XmlPullParser xpp = reader.getXPPParser();
         // Reset the parser since a new stream header has been sent from the client
-		if (connection.getTLSStreamHandler() == null)
-		{
-			xpp.setInput(new InputStreamReader(new ZipInputStream (socket.getInputStream()), CHARSET));
+        if (connection.getTLSStreamHandler() == null)
+        {
+            ZInputStream in = new ZInputStream(socket.getInputStream());
+            in.setFlushMode(JZlib.Z_PARTIAL_FLUSH);
+            xpp.setInput(new InputStreamReader(in, CHARSET));
         }
-		else
-		{
-			xpp.setInput(new InputStreamReader(new ZipInputStream (connection.getTLSStreamHandler().getInputStream()),
-                    CHARSET));
+        else
+        {
+            ZInputStream in = new ZInputStream(connection.getTLSStreamHandler().getInputStream());
+            in.setFlushMode(JZlib.Z_PARTIAL_FLUSH);
+            xpp.setInput(new InputStreamReader(in, CHARSET));
         }
 
         // Skip the opening stream sent by the client
-		for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;)
-		{
+        for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;)
+        {
             eventType = xpp.next();
         }
 
-		StringBuilder sb = new StringBuilder(340);
-		sb.append(geStreamHeader());
-		sb.append("<stream:features>");
+        StringBuilder sb = new StringBuilder(340);
+        sb.append(geStreamHeader());
+        sb.append("<stream:features>");
         // Include SASL mechanisms only if client has not been authenticated
         if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
             // Include available SASL Mechanisms
             sb.append(SASLAuthentication.getSASLMechanisms(session));
         }
-		// Include specific features such as resource binding and session establishment
-		// for client sessions
-		String specificFeatures = session.getAvailableStreamFeatures();
-		if (specificFeatures != null)
-		{
-			sb.append(specificFeatures);
-		}
-		sb.append("</stream:features>");
-		connection.deliverRawText(sb.toString());
+        // Include specific features such as resource binding and session establishment
+        // for client sessions
+        String specificFeatures = session.getAvailableStreamFeatures();
+        if (specificFeatures != null)
+        {
+            sb.append(specificFeatures);
+        }
+        sb.append("</stream:features>");
+        connection.deliverRawText(sb.toString());
     }
 
     /**