Author: timfox Date: 2009-08-21 16:41:47 -0400 (Fri, 21 Aug 2009) New Revision: 7878 Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java trunk/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionSendContinuationMessage.java trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java Log: fixed large message memory issue Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-08-21 17:16:34 UTC (rev 7877) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-08-21 20:41:47 UTC (rev 7878) @@ -15,6 +15,7 @@ import static org.hornetq.utils.SimpleString.toSimpleString; +import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; @@ -216,12 +217,12 @@ SessionSendMessage message = new SessionSendMessage(msg, sendBlocking); if (msg.getBodyInputStream() != null || msg.getEncodeSize() >= minLargeMessageSize || msg.isLargeMessage()) - { + { sendMessageInChunks(sendBlocking, msg); } else if (sendBlocking) - { - channel.sendBlocking(message); + { + channel.sendBlocking(message); } else { @@ -240,7 +241,7 @@ if (headerSize >= minLargeMessageSize) { throw new HornetQException(HornetQException.ILLEGAL_STATE, "Header size (" + headerSize + - ") is too big, use the messageBody for large data, or increase minLargeMessageSize"); + ") is too big, use the messageBody for large data, or increase minLargeMessageSize"); } // msg.getBody() could be Null on LargeServerMessage @@ -256,45 +257,67 @@ channel.send(initialChunk); - if (msg.getBodyInputStream() != null) + InputStream input = msg.getBodyInputStream(); + + if (input != null) { boolean lastChunk = false; - InputStream input = msg.getBodyInputStream(); + while (!lastChunk) { - byte[] bytesRead = new byte[minLargeMessageSize]; - int numberOfBytesRead; - - try - { - numberOfBytesRead = input.read(bytesRead); + byte[] buff = new byte[minLargeMessageSize]; + + int pos = 0; + + do + { + int numberOfBytesRead; + + int wanted = minLargeMessageSize - pos; + + try + { + numberOfBytesRead = input.read(buff, pos, wanted); + } + catch (IOException e) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, + "Error reading the LargeMessageBody", + e); + } + + if (numberOfBytesRead == -1) + { + lastChunk = true; + + break; + } + + pos += numberOfBytesRead; } - catch (IOException e) + while (pos < minLargeMessageSize); + + if (lastChunk) { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, - "Error reading the LargeMessageBody", - e); + byte[] buff2 = new byte[pos]; + + System.arraycopy(buff, 0, buff2, 0, pos); + + buff = buff2; } - - if (numberOfBytesRead < 0) - { - numberOfBytesRead = 0; - lastChunk = true; - } - - final SessionSendContinuationMessage chunk = new SessionSendContinuationMessage(bytesRead, - numberOfBytesRead, + + final SessionSendContinuationMessage chunk = new SessionSendContinuationMessage(buff, !lastChunk, lastChunk && sendBlocking); if (sendBlocking && lastChunk) { - // When sending it blocking, only the last chunk will be blocking. + // When sending it blocking, only the last chunk will be blocking. channel.sendBlocking(chunk); } else { - channel.send(chunk); + channel.send(chunk); } } @@ -305,8 +328,8 @@ catch (IOException e) { throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, - "Error closing stream from LargeMessageBody", - e); + "Error closing stream from LargeMessageBody", + e); } } else @@ -327,18 +350,17 @@ lastChunk = pos >= bodySize; - final SessionSendContinuationMessage chunk = new SessionSendContinuationMessage(bodyBuffer.array(), - chunkLength, + final SessionSendContinuationMessage chunk = new SessionSendContinuationMessage(bodyBuffer.array(), !lastChunk, lastChunk && sendBlocking); if (sendBlocking && lastChunk) { - // When sending it blocking, only the last chunk will be blocking. + // When sending it blocking, only the last chunk will be blocking. channel.sendBlocking(chunk); } else - { + { channel.send(chunk); } } Modified: trunk/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java 2009-08-21 17:16:34 UTC (rev 7877) +++ trunk/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java 2009-08-21 20:41:47 UTC (rev 7878) @@ -534,8 +534,6 @@ boolean attemptFailoverOrReconnect = (backupConnectorFactory != null || reconnectAttempts != 0) && (failoverOnServerShutdown || !serverShutdown); - log.info("Attempting failover or reconnect " + attemptFailoverOrReconnect); - if (attemptFailoverOrReconnect) { lockAllChannel1s(); @@ -640,8 +638,6 @@ } else { - log.info("Just closing connections and calling failure listeners"); - closeConnectionsAndCallFailureListeners(me); } } @@ -1096,7 +1092,6 @@ if (type == PacketImpl.DISCONNECT) { - log.info("Got a disconnect message"); threadPool.execute(new Runnable() { // Must be executed on new thread since cannot block the netty thread for a long time and fail can Modified: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionSendContinuationMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionSendContinuationMessage.java 2009-08-21 17:16:34 UTC (rev 7877) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionSendContinuationMessage.java 2009-08-21 20:41:47 UTC (rev 7878) @@ -35,9 +35,6 @@ private boolean requiresResponse; - // Not sent through the wire. Just to define how many bytes to send of body - private transient int bodyLength; - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -56,14 +53,12 @@ * @param continues * @param requiresResponse */ - public SessionSendContinuationMessage(final byte[] body, - final int bodyLength, + public SessionSendContinuationMessage(final byte[] body, final boolean continues, final boolean requiresResponse) { super(SESS_SEND_CONTINUATION, body, continues); this.requiresResponse = requiresResponse; - this.bodyLength = bodyLength; } @@ -80,25 +75,20 @@ @Override public int getRequiredBufferSize() { - return SESSION_CONTINUATION_BASE_SIZE + bodyLength + DataConstants.SIZE_BOOLEAN; + return super.getRequiredBufferSize() + DataConstants.SIZE_BOOLEAN; } @Override public void encodeBody(final HornetQBuffer buffer) { - buffer.writeInt(bodyLength); - buffer.writeBytes(body, 0, bodyLength); - buffer.writeBoolean(continues); + super.encodeBody(buffer); buffer.writeBoolean(requiresResponse); } @Override public void decodeBody(final HornetQBuffer buffer) { - bodyLength = buffer.readInt(); - body = new byte[bodyLength]; - buffer.readBytes(body); - continues = buffer.readBoolean(); + super.decodeBody(buffer); requiresResponse = buffer.readBoolean(); } Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2009-08-21 17:16:34 UTC (rev 7877) +++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2009-08-21 20:41:47 UTC (rev 7878) @@ -27,6 +27,7 @@ import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; import org.hornetq.core.client.impl.ClientConsumerInternal; +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.config.Configuration; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.Message; @@ -69,15 +70,17 @@ public void testDLALargeMessage() throws Exception { - final int messageSize = 50000; + final int messageSize = (int)(3.5 * ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE); - ClientSession session = null; + ClientSession session = null; try { server = createServer(true); server.start(); + + log.info("*********** starting test"); ClientSessionFactory sf = createInVMFactory(); @@ -101,6 +104,8 @@ Message clientFile = createLargeClientMessage(session, messageSize, true); + log.info("*********** sending large message"); + producer.send(clientFile); session.commit(); @@ -196,7 +201,7 @@ public void testDLAOnExpiry() throws Exception { - final int messageSize = 50000; + final int messageSize = (int)(3.5 * ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE); ClientSession session = null; @@ -332,7 +337,7 @@ public void testExpiryLargeMessage() throws Exception { - final int messageSize = 50000; + final int messageSize = (int)(3 * ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE); ClientSession session = null;