HORNETQ-448

Large Message Compression Impl
hornetq · Dec 5, 2010 · 167f003 · 167f003
1 parent 5806ee1
commit 167f003
Show file tree

Hide file tree

Showing 44 changed files with 2,627 additions and 119 deletions.
diff --git a/src/config/common/schema/hornetq-jms.xsd b/src/config/common/schema/hornetq-jms.xsd
@@ -75,7 +75,11 @@
             </xsd:element>
    			<xsd:element name="min-large-message-size" type="xsd:long"
    				maxOccurs="1" minOccurs="0">
-   			</xsd:element>   			          
+   			</xsd:element>   			 
+   			<xsd:element name="compress-large-messages" type="xsd:boolean"
+                maxOccurs="1" minOccurs="0">
+            </xsd:element>
+
             <xsd:element name="client-id" type="xsd:string"
                 maxOccurs="1" minOccurs="0">
             </xsd:element>            

diff --git a/src/main/org/hornetq/api/core/Message.java b/src/main/org/hornetq/api/core/Message.java
@@ -65,6 +65,8 @@ public interface Message
    public static final SimpleString HDR_ORIG_MESSAGE_ID = new SimpleString("_HQ_ORIG_MESSAGE_ID");
 
    public static final SimpleString HDR_GROUP_ID = new SimpleString("_HQ_GROUP_ID");
+
+   public static final SimpleString HDR_LARGE_COMPRESSED = new SimpleString("_HQ_LARGE_COMPRESSED");
 
    public static final SimpleString HDR_SCHEDULED_DELIVERY_TIME = new SimpleString("_HQ_SCHED_DELIVERY");
 

diff --git a/src/main/org/hornetq/api/core/client/ClientSessionFactory.java b/src/main/org/hornetq/api/core/client/ClientSessionFactory.java
@@ -138,4 +138,7 @@ ClientSession createSession(String username,
 
    CoreRemotingConnection getConnection();
 
+   void setCompressLargeMessages(boolean compressLargeMessage);
+
+   boolean isCompressLargeMessages();
 }
diff --git a/src/main/org/hornetq/api/core/client/HornetQClient.java b/src/main/org/hornetq/api/core/client/HornetQClient.java
@@ -43,6 +43,8 @@ public class HornetQClient
    // Any message beyond this size is considered a large message (to be sent in chunks)
 
    public static final int DEFAULT_MIN_LARGE_MESSAGE_SIZE = 100 * 1024;
+
+   public static final boolean DEFAULT_COMPRESS_LARGE_MESSAGES = false;
 
    public static final int DEFAULT_CONSUMER_WINDOW_SIZE = 1024 * 1024;
 

diff --git a/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java b/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
@@ -29,6 +29,7 @@
 import org.hornetq.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage;
 import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage;
 import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveLargeMessage;
+import org.hornetq.utils.DecompressedLargeMessageBuffer;
 import org.hornetq.utils.Future;
 import org.hornetq.utils.PriorityLinkedList;
 import org.hornetq.utils.PriorityLinkedListImpl;
@@ -454,6 +455,11 @@ public Exception getLastException()
    // ClientConsumerInternal implementation
    // --------------------------------------------------------------
 
+   public ClientSessionInternal getSession()
+   {
+      return session;
+   }
+
    public SessionQueueQueryResponseMessage getQueueInfo()
    {
       return queueInfo;
@@ -554,7 +560,14 @@ public synchronized void handleLargeMessage(final SessionReceiveLargeMessage pac
 
       currentLargeMessageBuffer = new LargeMessageBufferImpl(this, packet.getLargeMessageSize(), 60, largeMessageCache);
 
-      currentChunkMessage.setBuffer(currentLargeMessageBuffer);
+      if (currentChunkMessage.isCompressed())
+      {
+         currentChunkMessage.setBuffer(new DecompressedLargeMessageBuffer(currentLargeMessageBuffer));
+      }
+      else
+      {
+         currentChunkMessage.setBuffer(currentLargeMessageBuffer);
+      }
 
       currentChunkMessage.setFlowControlSize(0);
 

diff --git a/src/main/org/hornetq/core/client/impl/ClientConsumerInternal.java b/src/main/org/hornetq/core/client/impl/ClientConsumerInternal.java
@@ -67,4 +67,6 @@ public interface ClientConsumerInternal extends ClientConsumer
    void start();
 
    SessionQueueQueryResponseMessage getQueueInfo();
+
+   ClientSessionInternal getSession();
 }
diff --git a/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java b/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java
@@ -21,11 +21,11 @@
 import org.hornetq.api.core.HornetQBuffer;
 import org.hornetq.api.core.HornetQBuffers;
 import org.hornetq.api.core.HornetQException;
+import org.hornetq.api.core.Message;
 import org.hornetq.api.core.SimpleString;
 import org.hornetq.core.logging.Logger;
 import org.hornetq.core.message.BodyEncoder;
 import org.hornetq.core.message.impl.MessageImpl;
-import org.hornetq.utils.DataConstants;
 
 /**
  * 
@@ -117,6 +117,11 @@ public boolean isLargeMessage()
    {
       return largeMessage;
    }
+
+   public boolean isCompressed()
+   {
+      return properties.getBooleanProperty(Message.HDR_LARGE_COMPRESSED);
+   }
 
    /**
     * @param largeMessage the largeMessage to set
@@ -142,15 +147,14 @@ public String toString()
              "]";
    }
 
-   // FIXME - only used for large messages - move it!
    /* (non-Javadoc)
     * @see org.hornetq.api.core.client.ClientMessage#saveToOutputStream(java.io.OutputStream)
     */
    public void saveToOutputStream(final OutputStream out) throws HornetQException
    {
       if (largeMessage)
       {
-         ((LargeMessageBufferInternal)getWholeBuffer()).saveBuffer(out);
+          ((LargeMessageBufferInternal)getWholeBuffer()).saveBuffer(out);
       }
       else
       {

diff --git a/src/main/org/hornetq/core/client/impl/ClientMessageInternal.java b/src/main/org/hornetq/core/client/impl/ClientMessageInternal.java
@@ -44,4 +44,6 @@ public interface ClientMessageInternal extends ClientMessage, MessageInternal
    void discardLargeBody();
 
    void setBuffer(HornetQBuffer buffer);
+
+   boolean isCompressed();
 }
diff --git a/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java b/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
@@ -22,12 +22,13 @@
 import org.hornetq.api.core.Message;
 import org.hornetq.api.core.SimpleString;
 import org.hornetq.core.logging.Logger;
-import org.hornetq.core.message.BodyEncoder;
 import org.hornetq.core.message.impl.MessageInternal;
 import org.hornetq.core.protocol.core.Channel;
 import org.hornetq.core.protocol.core.impl.wireformat.SessionSendContinuationMessage;
 import org.hornetq.core.protocol.core.impl.wireformat.SessionSendLargeMessage;
 import org.hornetq.core.protocol.core.impl.wireformat.SessionSendMessage;
+import org.hornetq.utils.DeflaterReader;
+import org.hornetq.utils.HornetQBufferInputStream;
 import org.hornetq.utils.TokenBucketLimiter;
 import org.hornetq.utils.UUIDGenerator;
 
@@ -150,7 +151,7 @@ public synchronized void close() throws HornetQException
       {
          return;
       }
-            
+
       doCleanup();
    }
 
@@ -190,7 +191,7 @@ public ClientProducerCredits getProducerCredits()
    {
       return credits;
    }
-   
+
    // Protected ------------------------------------------------------------------------------------
 
    // Package Private ------------------------------------------------------------------------------
@@ -203,7 +204,7 @@ private void doCleanup()
       {
          session.returnCredits(address);
       }
-      
+
       session.removeProducer(this);
 
       closed = true;
@@ -212,12 +213,13 @@ private void doCleanup()
    private void doSend(final SimpleString address, final Message msg) throws HornetQException
    {
       MessageInternal msgI = (MessageInternal)msg;
-      
+
       ClientProducerCredits theCredits;
-      
+
       boolean isLarge;
 
-      if (msgI.getBodyInputStream() != null || msgI.isLargeMessage() || msgI.getBodyBuffer().writerIndex() > minLargeMessageSize)
+      if (msgI.getBodyInputStream() != null || msgI.isLargeMessage() ||
+          msgI.getBodyBuffer().writerIndex() > minLargeMessageSize)
       {
          isLarge = true;
       }
@@ -236,7 +238,7 @@ private void doSend(final SimpleString address, final Message msg) throws Hornet
          {
             msg.setAddress(address);
          }
-         
+
          // Anonymous
          theCredits = session.getCredits(address, true);
       }
@@ -250,7 +252,7 @@ private void doSend(final SimpleString address, final Message msg) throws Hornet
          {
             msg.setAddress(this.address);
          }
-         
+
          theCredits = credits;
       }
 
@@ -270,8 +272,6 @@ private void doSend(final SimpleString address, final Message msg) throws Hornet
 
       session.workDone();
 
-
-
       if (isLarge)
       {
          largeMessageSend(sendBlocking, msgI, theCredits);
@@ -322,8 +322,16 @@ private void checkClosed() throws HornetQException
     * @param msgI
     * @throws HornetQException
     */
-   private void largeMessageSend(final boolean sendBlocking, final MessageInternal msgI, final ClientProducerCredits credits) throws HornetQException
+   private void largeMessageSend(final boolean sendBlocking,
+                                 final MessageInternal msgI,
+                                 final ClientProducerCredits credits) throws HornetQException
    {
+
+      if (session.isCompressLargeMessages())
+      {
+         msgI.putBooleanProperty(Message.HDR_LARGE_COMPRESSED, true);
+      }
+
       int headerSize = msgI.getHeadersAndPropertiesEncodeSize();
 
       if (headerSize >= minLargeMessageSize)
@@ -341,7 +349,6 @@ private void largeMessageSend(final boolean sendBlocking, final MessageInternal
       HornetQBuffer headerBuffer = HornetQBuffers.fixedBuffer(headerSize);
 
       msgI.encodeHeadersAndProperties(headerBuffer);
-
       SessionSendLargeMessage initialChunk = new SessionSendLargeMessage(headerBuffer.toByteBuffer().array());
 
       channel.send(initialChunk);
@@ -358,7 +365,7 @@ private void largeMessageSend(final boolean sendBlocking, final MessageInternal
 
       if (input != null)
       {
-         largeMessageSendStreamed(sendBlocking, input, credits);
+         largeMessageSendStreamed(sendBlocking, msgI, input, credits);
       }
       else
       {
@@ -375,72 +382,29 @@ private void largeMessageSendBuffered(final boolean sendBlocking,
                                          final MessageInternal msgI,
                                          final ClientProducerCredits credits) throws HornetQException
    {
-      BodyEncoder context = msgI.getBodyEncoder();
-
-      final long bodySize = context.getLargeBodySize();
-
-      context.open();
-      try
-      {
-
-         for (int pos = 0; pos < bodySize;)
-         {
-            final boolean lastChunk;
-
-            final int chunkLength = Math.min((int)(bodySize - pos), minLargeMessageSize);
-
-            final HornetQBuffer bodyBuffer = HornetQBuffers.fixedBuffer(chunkLength);
-
-            context.encode(bodyBuffer, chunkLength);
-
-            pos += chunkLength;
-
-            lastChunk = pos >= bodySize;
-
-            final SessionSendContinuationMessage chunk = new SessionSendContinuationMessage(bodyBuffer.toByteBuffer()
-                                                                                                      .array(),
-                                                                                            !lastChunk,
-                                                                                            lastChunk && sendBlocking);
-
-            if (sendBlocking && lastChunk)
-            {
-               // When sending it blocking, only the last chunk will be blocking.
-               channel.sendBlocking(chunk);
-            }
-            else
-            {
-               channel.send(chunk);
-            }
-
-            try
-            {
-               credits.acquireCredits(chunk.getPacketSize());
-            }
-            catch (InterruptedException e)
-            {
-            }
-         }
-      }
-      finally
-      {
-         context.close();
-      }
+      msgI.getBodyBuffer().readerIndex(0);
+      largeMessageSendStreamed(sendBlocking, msgI, new HornetQBufferInputStream(msgI.getBodyBuffer()), credits);
    }
 
    /**
-    * TODO: This method could be eliminated and 
-    *       combined with {@link ClientProducerImpl#largeMessageSendBuffered(boolean, Message, ClientProducerCredits)}. 
-    *       All that's needed for this is ClientMessage returning the proper BodyEncoder for streamed
     * @param sendBlocking
     * @param input
     * @throws HornetQException
     */
    private void largeMessageSendStreamed(final boolean sendBlocking,
-                                         final InputStream input,
+                                         final MessageInternal msgI,
+                                         final InputStream inputStreamParameter,
                                          final ClientProducerCredits credits) throws HornetQException
    {
       boolean lastPacket = false;
 
+      InputStream input = inputStreamParameter;
+
+      if (session.isCompressLargeMessages())
+      {
+         input = new DeflaterReader(inputStreamParameter);
+      }
+
       while (!lastPacket)
       {
          byte[] buff = new byte[minLargeMessageSize];

diff --git a/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java b/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
@@ -30,6 +30,7 @@
 
 import org.hornetq.api.core.*;
 import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.api.core.client.HornetQClient;
 import org.hornetq.api.core.client.ServerLocator;
 import org.hornetq.api.core.client.SessionFailureListener;
 import org.hornetq.core.logging.Logger;
@@ -146,6 +147,8 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
    public final Exception e = new Exception();
 
    private final Object waitLock = new Object();
+
+   private boolean compressLargeMessages;
 
    // Static
    // ---------------------------------------------------------------------------------------
@@ -202,6 +205,8 @@ public ClientSessionFactoryImpl(final ServerLocatorInternal serverLocator,
       closeExecutor = orderedExecutorFactory.getExecutor();
 
       this.interceptors = interceptors;
+
+      compressLargeMessages = HornetQClient.DEFAULT_COMPRESS_LARGE_MESSAGES;
 
    }
 
@@ -768,6 +773,7 @@ private ClientSession createSessionInternal(final String username,
                                                                      serverLocator.isBlockOnDurableSend(),
                                                                      serverLocator.isCacheLargeMessagesClient(),
                                                                      serverLocator.getMinLargeMessageSize(),
+                                                                     compressLargeMessages,
                                                                      serverLocator.getInitialMessagePacketSize(),
                                                                      serverLocator.getGroupID(),
                                                                      connection,
@@ -1358,4 +1364,14 @@ public synchronized void cancel()
          cancelled = true;
       }
    }
+
+   public void setCompressLargeMessages(boolean compressLargeMessage)
+   {
+      this.compressLargeMessages = compressLargeMessage;
+   }
+
+   public boolean isCompressLargeMessages()
+   {
+      return this.compressLargeMessages;
+   }
 }