[hornetq-commits] JBoss hornetq SVN: r10113 - in trunk/src/main/org/hornetq: utils and 1 other directory.
Author: clebert.suconic(a)jboss.com
Date: 2011-01-10 17:52:30 -0500 (Mon, 10 Jan 2011)
New Revision: 10113
Modified:
trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
trunk/src/main/org/hornetq/utils/DeflaterReader.java
Log:
fixing tests
Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2011-01-10
20:30:22 UTC (rev 10112)
+++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2011-01-10
22:52:30 UTC (rev 10113)
@@ -468,10 +468,11 @@
// We won't know the real size of the message since we are compressing while
reading the streaming.
// This counter will be passed to the deflater to be updated for every byte read
+ AtomicLong messageSize = new AtomicLong();
if (session.isCompressLargeMessages())
{
- input = new DeflaterReader(inputStreamParameter);
+ input = new DeflaterReader(inputStreamParameter, messageSize);
}
long totalSize = 0;
@@ -517,13 +518,18 @@
if (lastPacket)
{
+ if (!session.isCompressLargeMessages())
+ {
+ messageSize.set(totalSize);
+ }
+
byte[] buff2 = new byte[pos];
System.arraycopy(buff, 0, buff2, 0, pos);
buff = buff2;
- chunk = new SessionSendContinuationMessage(buff, false, sendBlocking,
totalSize);
+ chunk = new SessionSendContinuationMessage(buff, false, sendBlocking,
messageSize.get());
}
else
{
Modified: trunk/src/main/org/hornetq/utils/DeflaterReader.java
===================================================================
--- trunk/src/main/org/hornetq/utils/DeflaterReader.java 2011-01-10 20:30:22 UTC (rev
10112)
+++ trunk/src/main/org/hornetq/utils/DeflaterReader.java 2011-01-10 22:52:30 UTC (rev
10113)
@@ -41,11 +41,6 @@
input = inData;
this.bytesRead = bytesRead;
}
-
- public DeflaterReader(final InputStream inData)
- {
- this(inData, null);
- }
@Override
public int read() throws IOException