2:42 a.m.
Author: timfox
Date: 2009-11-05 03:42:21 -0500 (Thu, 05 Nov 2009)
New Revision: 8221
Modified:
trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java
trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
Log:
https://jira.jboss.org/jira/browse/HORNETQ-206
Modified:
trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java
===================================================================
---
trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java 2009-11-05
07:56:15 UTC (rev 8220)
+++
trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java 2009-11-05
08:42:21 UTC (rev 8221)
@@ -47,7 +47,9 @@
// The message we will send is size 256MB, even though we are only running in 50MB of
RAM on both client and server.
// HornetQ will support much larger message sizes, but we use 512MB so the example
runs in reasonable time.
- private final long FILE_SIZE = 256 * 1024 * 1024;
+ // private final long FILE_SIZE = 256L * 1024 * 1024;
+
+ private final long FILE_SIZE = 2L * 1024 * 1024 * 1024; // 2 GiB message
public boolean runExample() throws Exception
{
Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-11-05
07:56:15 UTC (rev 8220)
+++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-11-05
08:42:21 UTC (rev 8221)
@@ -22,8 +22,8 @@
import org.hornetq.core.buffers.ChannelBuffers;
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.logging.Logger;
+import org.hornetq.core.message.LargeMessageEncodingContext;
import org.hornetq.core.message.Message;
-import org.hornetq.core.message.LargeMessageEncodingContext;
import org.hornetq.core.message.impl.MessageImpl;
import org.hornetq.core.remoting.Channel;
import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage;
@@ -250,7 +250,7 @@
if (isLarge)
{
- largeMessageSend(sendBlocking, msg);
+ largeMessageSend(sendBlocking, msg, theCredits);
}
else if (sendBlocking)
{
@@ -269,14 +269,8 @@
// Not the continuations, but this is ok since we are only interested in
limiting the amount of
// data in *memory* and continuations go straight to the disk
- if (isLarge)
+ if (!isLarge)
{
- // TODO this is pretty hacky - we should define consistent meanings of encode
size
-
- theCredits.acquireCredits(msg.getHeadersAndPropertiesEncodeSize());
- }
- else
- {
theCredits.acquireCredits(msg.getEncodeSize());
}
}
@@ -292,15 +286,14 @@
throw new HornetQException(HornetQException.OBJECT_CLOSED, "Producer is
closed");
}
}
-
-
+
// Methods to send Large
Messages----------------------------------------------------------------
-
+
/**
* @param msg
* @throws HornetQException
*/
- private void largeMessageSend(final boolean sendBlocking, final Message msg) throws
HornetQException
+ private void largeMessageSend(final boolean sendBlocking, final Message msg, final
ClientProducerCredits credits) throws HornetQException
{
int headerSize = msg.getHeadersAndPropertiesEncodeSize();
@@ -323,15 +316,23 @@
channel.send(initialChunk);
+ try
+ {
+ credits.acquireCredits(msg.getHeadersAndPropertiesEncodeSize());
+ }
+ catch (InterruptedException e)
+ {
+ }
+
InputStream input = msg.getBodyInputStream();
if (input != null)
{
- largeMessageSendStreamed(sendBlocking, input);
+ largeMessageSendStreamed(sendBlocking, input, credits);
}
else
{
- largeMessageSendBuffered(sendBlocking, msg);
+ largeMessageSendBuffered(sendBlocking, msg, credits);
}
}
@@ -340,7 +341,9 @@
* @param msg
* @throws HornetQException
*/
- private void largeMessageSendBuffered(final boolean sendBlocking, final Message msg)
throws HornetQException
+ private void largeMessageSendBuffered(final boolean sendBlocking,
+ final Message msg,
+ final ClientProducerCredits credits) throws
HornetQException
{
final long bodySize = msg.getLargeBodySize();
@@ -373,6 +376,14 @@
{
channel.send(chunk);
}
+
+ try
+ {
+ credits.acquireCredits(chunk.getRequiredBufferSize());
+ }
+ catch (InterruptedException e)
+ {
+ }
}
}
@@ -381,7 +392,9 @@
* @param input
* @throws HornetQException
*/
- private void largeMessageSendStreamed(final boolean sendBlocking, InputStream input)
throws HornetQException
+ private void largeMessageSendStreamed(final boolean sendBlocking,
+ final InputStream input,
+ final ClientProducerCredits credits) throws
HornetQException
{
boolean lastPacket = false;
@@ -441,6 +454,14 @@
{
channel.send(chunk);
}
+
+ try
+ {
+ credits.acquireCredits(chunk.getRequiredBufferSize());
+ }
+ catch (InterruptedException e)
+ {
+ }
}
try
@@ -455,8 +476,6 @@
}
}
-
-
// Inner Classes
--------------------------------------------------------------------------------
class DecodingContext implements LargeMessageEncodingContext
{
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-11-05 07:56:15
UTC (rev 8220)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-11-05 08:42:21
UTC (rev 8221)
@@ -1478,7 +1478,7 @@
{
try
{
- releaseOutStanding(message);
+ releaseOutStanding(message, message.getEncodeSize());
}
catch (Exception e)
{
@@ -1499,20 +1499,23 @@
{
throw new HornetQException(HornetQException.ILLEGAL_STATE,
"large-message not initialized on server");
}
-
+
+ //Immediately release the credits for the continuations- these don't
contrinute to the in-memory size
+ //of the message
+
+ releaseOutStanding(currentLargeMessage, packet.getRequiredBufferSize());
+
currentLargeMessage.addBytes(packet.getBody());
if (!packet.isContinues())
- {
- final LargeServerMessage message = currentLargeMessage;
+ {
+ currentLargeMessage.releaseResources();
+ send(currentLargeMessage);
+
+ releaseOutStanding(currentLargeMessage,
currentLargeMessage.getEncodeSize());
+
currentLargeMessage = null;
-
- message.releaseResources();
-
- send(message);
-
- releaseOutStanding(message);
}
if (packet.isRequiresResponse())
@@ -1910,17 +1913,15 @@
* returned. When a session closes any outstanding credits will be returned.
*
*/
- private void releaseOutStanding(final ServerMessage message) throws Exception
+ private void releaseOutStanding(final ServerMessage message, final int credits) throws
Exception
{
CreditManagerHolder holder = getCreditManagerHolder(message.getDestination());
- int size = message.getEncodeSize();
+ holder.outstandingCredits -= credits;
- holder.outstandingCredits -= size;
-
- holder.store.returnProducerCredits(size);
+ holder.store.returnProducerCredits(credits);
}
-
+
private void send(final ServerMessage msg) throws Exception
{
// check the user has write access to this address.