diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java index d6daac6a2e..7937f7fc36 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java @@ -19,6 +19,7 @@ package org.apache.activemq.artemis.protocol.amqp.broker; import java.nio.ByteBuffer; +import io.netty.buffer.ByteBuf; import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.api.core.ICoreMessage; @@ -74,6 +75,12 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage } } } + + /** + * AMQPLargeMessagePersister will save the buffer here. + * */ + volatile ByteBuf temporaryBuffer; + private final LargeBody largeBody; /** * We control durability on a separate property here, as we need to know if it's durable ahead of the header parsing. diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java index bdafcba0b1..cffda5de23 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java @@ -34,11 +34,6 @@ import static org.apache.activemq.artemis.core.persistence.PersisterIDs.AMQPLarg public class AMQPLargeMessagePersister extends MessagePersister { private static final Logger log = Logger.getLogger(AMQPLargeMessagePersister.class); - // We need to save the encoder ahead of time - // as we need to know the exact size of the Encoding - // so we store the savedBuffer on the getEncodeSize before we actually store it - private static final ThreadLocal savedBuffer = new ThreadLocal<>(); - public static final byte ID = AMQPLargeMessagePersister_ID; public static AMQPLargeMessagePersister theInstance; @@ -72,14 +67,12 @@ public class AMQPLargeMessagePersister extends MessagePersister { } private ByteBuf getSavedEncodeBuffer(Message record) { - ByteBuf buf = savedBuffer.get(); - if (buf == null) { - AMQPLargeMessage largeMessage = (AMQPLargeMessage)record; - buf = PooledByteBufAllocator.DEFAULT.buffer(largeMessage.getEstimateSavedEncode()); - largeMessage.saveEncoding(buf); - savedBuffer.set(buf); + AMQPLargeMessage largeMessage = (AMQPLargeMessage)record; + if (largeMessage.temporaryBuffer == null) { + largeMessage.temporaryBuffer = PooledByteBufAllocator.DEFAULT.buffer(largeMessage.getEstimateSavedEncode()); + largeMessage.saveEncoding(largeMessage.temporaryBuffer); } - return buf; + return largeMessage.temporaryBuffer; } /** @@ -89,7 +82,7 @@ public class AMQPLargeMessagePersister extends MessagePersister { public void encode(ActiveMQBuffer buffer, Message record) { super.encode(buffer, record); - AMQPMessage msgEncode = (AMQPMessage) record; + AMQPLargeMessage msgEncode = (AMQPLargeMessage) record; buffer.writeLong(record.getMessageID()); buffer.writeBoolean(record.isDurable()); @@ -106,8 +99,7 @@ public class AMQPLargeMessagePersister extends MessagePersister { ByteBuf savedEncodeBuffer = getSavedEncodeBuffer(record); buffer.writeBytes(savedEncodeBuffer, 0, savedEncodeBuffer.writerIndex()); savedEncodeBuffer.release(); - - savedBuffer.set(null); + msgEncode.temporaryBuffer = null; } @Override diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java index 32879747fb..3ab3f21b92 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java @@ -255,8 +255,8 @@ public final class LargeServerMessageImpl extends CoreMessage implements CoreLar @Override public Message copy() { SequentialFile newfile = storageManager.createFileForLargeMessage(messageID, durable); - - Message newMessage = new LargeServerMessageImpl(this, properties, newfile, messageID); + LargeServerMessageImpl newMessage = new LargeServerMessageImpl(this, properties, newfile, messageID); + newMessage.setParentRef(this); return newMessage; } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java index c7493268dc..2539838ec2 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java @@ -53,28 +53,30 @@ import org.junit.runners.Parameterized; */ @RunWith(value = Parameterized.class) public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { + private String smallFrameAcceptor = new String("tcp://localhost:" + (AMQP_PORT + 8)); int frameSize; + int minLargeMessageSize; - - - @Parameterized.Parameters(name = "frameSize = {0}") - public static Iterable persistenceEnabled() { + @Parameterized.Parameters(name = "frameSize = {0}, minLargeMessage = {1}") + public static Iterable testParameters() { // The reason I use two frames sizes here // is because a message that wasn't broken into frames // but still beyond 50K, should still be considered large when storing - return Arrays.asList(new Object[][]{{512}, {1024 * 1024}}); + return Arrays.asList(new Object[][]{{512, 50000}, {1024 * 1024, 50000}, + // we disable large message for at least one parameter to compare results between large and non large messages + {1024 * 1024, 50000000}}); } - public SimpleStreamingLargeMessageTest(int frameSize) { + public SimpleStreamingLargeMessageTest(int frameSize, int minLargeMessageSize) { this.frameSize = frameSize; + this.minLargeMessageSize = minLargeMessageSize; } - @Override protected void addAdditionalAcceptors(ActiveMQServer server) throws Exception { - server.getConfiguration().addAcceptorConfiguration("flow", smallFrameAcceptor + "?protocols=AMQP;useEpoll=false;maxFrameSize=" + frameSize + ";amqpMinLargeMessageSize=50000"); + server.getConfiguration().addAcceptorConfiguration("flow", smallFrameAcceptor + "?protocols=AMQP;useEpoll=false;maxFrameSize=" + frameSize + ";amqpMinLargeMessageSize=" + minLargeMessageSize); } @Test(timeout = 60000) @@ -143,10 +145,10 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { for (int i = 0; i < 10; i++) { AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS); Assert.assertNotNull(msgReceived); - Data body = (Data)msgReceived.getWrappedMessage().getBody(); + Data body = (Data) msgReceived.getWrappedMessage().getBody(); byte[] bodyArray = body.getValue().getArray(); for (int bI = 0; bI < size; bI++) { - Assert.assertEquals((byte)'z', bodyArray[bI]); + Assert.assertEquals((byte) 'z', bodyArray[bI]); } msgReceived.accept(true); } @@ -154,7 +156,6 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { receiver.flow(1); Assert.assertNull(receiver.receiveNoWait()); - receiver.close(); connection.close(); @@ -181,7 +182,7 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { @Test public void testSendWithPropertiesNonPersistent() throws Exception { - testSendWithPropertiesAndFilter(true, true); + testSendWithPropertiesAndFilter(false, false); } @@ -209,6 +210,8 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { AmqpMessage message = new AmqpMessage(); message.setDurable(persistent); boolean odd = (m % 2 == 0); + message.setApplicationProperty("i", m); + message.setApplicationProperty("oddString", odd ? "odd" : "even"); message.setApplicationProperty("odd", odd); if (odd) { message.setApplicationProperty("oddID", oddID++); @@ -245,14 +248,14 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { AmqpReceiver receiver = session.createReceiver(getQueueName(), "odd=true"); receiver.flow(10); for (int i = 0; i < 5; i++) { - System.out.println("Receiving " + i); AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS); Assert.assertNotNull(msgReceived); - System.out.println("Received " + msgReceived); - Data body = (Data)msgReceived.getWrappedMessage().getBody(); + Assert.assertTrue((boolean)msgReceived.getApplicationProperty("odd")); + Assert.assertEquals(i, (int)msgReceived.getApplicationProperty("oddID")); + Data body = (Data) msgReceived.getWrappedMessage().getBody(); byte[] bodyArray = body.getValue().getArray(); for (int bI = 0; bI < size; bI++) { - Assert.assertEquals((byte)'z', bodyArray[bI]); + Assert.assertEquals((byte) 'z', bodyArray[bI]); } msgReceived.accept(true); } @@ -260,9 +263,7 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { receiver.flow(1); Assert.assertNull(receiver.receiveNoWait()); - receiver.close(); - connection.close(); validateNoFilesOnLargeDir(getLargeMessagesDir(), 5); @@ -273,7 +274,6 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { } - @Test public void testJMSPersistentTX() throws Exception {