This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-jms.git
The following commit(s) were added to refs/heads/main by this push:
new 78781ea6 QPIDJMS-632: update to proton-j 0.35.0
78781ea6 is described below
commit 78781ea6d867ebb4185c8f786f6a873885983343
Author: Robbie Gemmell <[email protected]>
AuthorDate: Fri Jul 31 19:02:32 2026 +0100
QPIDJMS-632: update to proton-j 0.35.0
add options to configure new settings
---
pom.xml | 2 +-
.../qpid/jms/provider/amqp/AmqpConnection.java | 8 +
.../qpid/jms/provider/amqp/AmqpConsumer.java | 8 +
.../qpid/jms/provider/amqp/AmqpProvider.java | 56 +++++++
.../qpid/jms/provider/amqp/message/AmqpCodec.java | 10 +-
.../amqp/message/AmqpTypedObjectDelegate.java | 4 +-
.../jms/integration/ConnectionIntegrationTest.java | 16 +-
.../jms/integration/ConsumerIntegrationTest.java | 185 +++++++++++++++++++++
.../MultiTransferFrameMessageIntegrationTest.java | 61 +++++++
.../integration/ObjectMessageIntegrationTest.java | 4 +
.../jms/provider/amqp/AmqpProviderFactoryTest.java | 36 ++++
.../qpid/jms/provider/amqp/AmqpProviderTest.java | 52 ++++++
.../jms/provider/amqp/message/AmqpCodecTest.java | 131 ++++++++++++++-
.../amqp/message/AmqpJmsMessageTypesTestCase.java | 8 +
.../qpid/jms/test/testpeer/TestAmqpPeer.java | 12 +-
15 files changed, 578 insertions(+), 15 deletions(-)
diff --git a/pom.xml b/pom.xml
index e90f33f1..2ad68904 100644
--- a/pom.xml
+++ b/pom.xml
@@ -36,7 +36,7 @@
<properties>
<!-- Dependency Versions for this Project -->
- <proton-version>0.34.1</proton-version>
+ <proton-version>0.35.0</proton-version>
<netty-version>4.1.137.Final</netty-version>
<slf4j-version>2.0.18</slf4j-version>
<jakarta.jms-api-version>3.1.0</jakarta.jms-api-version>
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnection.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnection.java
index 6bdca103..ac7025cf 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnection.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnection.java
@@ -233,6 +233,14 @@ public class AmqpConnection extends
AmqpAbstractResource<JmsConnectionInfo, Conn
return getProvider().getAnonymousFallbackCacheTimeout();
}
+ public int getMaxDecodeDepth() {
+ return getProvider().getMaxDecodeDepth();
+ }
+
+ public int getZeroWidthArrayElementLimit() {
+ return getProvider().getZeroWidthArrayElementLimit();
+ }
+
/**
* @return the AMQP based JmsMessageFactory for this Connection.
*/
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
index 696a6984..163e2d4d 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
@@ -637,6 +637,14 @@ public class AmqpConsumer extends
AmqpAbstractResource<JmsConsumerInfo, Receiver
return session.getProvider().getDrainTimeout();
}
+ public int getMaxDecodeDepth() {
+ return session.getProvider().getMaxDecodeDepth();
+ }
+
+ public int getZeroWidthArrayElementLimit() {
+ return session.getProvider().getZeroWidthArrayElementLimit();
+ }
+
@Override
public String toString() {
return "AmqpConsumer { " + getResourceInfo().getId() + " }";
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
index d7f2f354..cd47e110 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
@@ -124,6 +124,9 @@ public class AmqpProvider implements Provider,
TransportListener , AmqpResourceP
private static final int DEFAULT_MAX_WRITE_BYTES_BEFORE_FLUSH = 128 * 1024;
private static final int DEFAULT_ANONYMOUS_FALLBACK_CACHE_TIMEOUT = 30000;
private static final int DEFAULT_ANONYMOUS_FALLBACK_CACHE_SIZE = 1;
+ private static final int DEFAULT_MAX_TRANSFERS_PER_DELIVERY = 65535;
+ private static final int DEFAULT_MAX_DECODE_DEPTH = 32;
+ private static final int DEFAULT_ZERO_WIDTH_ARRAY_ELEMENT_LIMIT = 0;
private volatile ProviderListener listener;
private volatile AmqpConnection connection;
@@ -144,6 +147,9 @@ public class AmqpProvider implements Provider,
TransportListener , AmqpResourceP
private int maxWriteBytesBeforeFlush =
DEFAULT_MAX_WRITE_BYTES_BEFORE_FLUSH;
private int anonymousFallbackCacheTimeout =
DEFAULT_ANONYMOUS_FALLBACK_CACHE_TIMEOUT;
private int anonymousFallbackCacheSize =
DEFAULT_ANONYMOUS_FALLBACK_CACHE_SIZE;
+ private int maxTransfersPerDelivery = DEFAULT_MAX_TRANSFERS_PER_DELIVERY;
+ private int maxDecodeDepth = DEFAULT_MAX_DECODE_DEPTH;
+ private int zeroWidthArrayElementLimit =
DEFAULT_ZERO_WIDTH_ARRAY_ELEMENT_LIMIT;
private boolean allowNonSecureRedirects;
@@ -246,6 +252,10 @@ public class AmqpProvider implements Provider,
TransportListener , AmqpResourceP
protonTransport.setOutboundFrameSizeLimit(getMaxFrameSize());
}
+ if (getMaxTransfersPerDelivery() > 0) {
+
protonTransport.setMaxTransfersPerDelivery(getMaxTransfersPerDelivery());
+ }
+
protonTransport.setChannelMax(getChannelMax());
protonTransport.setIdleTimeout(idleTimeout);
protonTransport.bind(protonConnection);
@@ -1389,6 +1399,52 @@ public class AmqpProvider implements Provider,
TransportListener , AmqpResourceP
this.maxFrameSize = maxFrameSize;
}
+ public int getZeroWidthArrayElementLimit() {
+ return zeroWidthArrayElementLimit;
+ }
+
+ /**
+ * Configures the maximum number of array elements allowed while decoding
arrays
+ * of zero-width elements (ulong0, uint0, list0, boolean-true,
boolean-false, null)
+ * before a decode exception.
+ *
+ * Default is 0, meaning only the empty array is permitted.
+ *
+ * @param zeroWidthArrayElementLimit The configured max zero-width
elements allowed.
+ */
+ public void setZeroWidthArrayElementLimit(int zeroWidthArrayElementLimit) {
+ this.zeroWidthArrayElementLimit = zeroWidthArrayElementLimit;
+ }
+
+ public int getMaxDecodeDepth() {
+ return maxDecodeDepth;
+ }
+
+ /**
+ * Configures the maximum decode depth allowed before error when decoding
Maps, Lists,
+ * Arrays, and certain related DescribedTypes from an encoded payload.
Default is 32.
+ *
+ * @param maxDecodeDepth The configured max decode depth allowed.
+ */
+ public void setMaxDecodeDepth(int maxDecodeDepth) {
+ this.maxDecodeDepth = maxDecodeDepth;
+ }
+
+ public int getMaxTransfersPerDelivery() {
+ return maxTransfersPerDelivery;
+ }
+
+ /**
+ * Sets the maximum number of transfer frames that can be used for an
incoming delivery.
+ *
+ * A value of -1 indicates that the proton default should be used.
+ *
+ * @param maxTransfersPerDelivery the maximum number of transfer frames
per incoming delivery.
+ */
+ public void setMaxTransfersPerDelivery(int maxTransfersPerDelivery) {
+ this.maxTransfersPerDelivery = maxTransfersPerDelivery;
+ }
+
public long getSessionOutgoingWindow() {
return sessionOutoingWindow;
}
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java
index 7a75d25f..0750335a 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java
@@ -36,6 +36,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import org.apache.qpid.jms.provider.amqp.AmqpConnection;
import org.apache.qpid.jms.provider.amqp.AmqpConsumer;
import org.apache.qpid.jms.util.ContentTypeSupport;
import org.apache.qpid.jms.util.InvalidContentTypeException;
@@ -130,18 +131,23 @@ public final class AmqpCodec {
/**
* Given an encoded AMQP Section, decode the value previously written
there.
*
+ * @param connection
+ * the connection this is being done for
* @param encoded
* the AMQP Section value to decode.
*
* @return a Section object read from its encoded form.
*/
- public static Section decode(ByteBuf encoded) {
+ public static Section decode(AmqpConnection connection, ByteBuf encoded) {
if (encoded == null || !encoded.isReadable()) {
return null;
}
DecoderImpl decoder = TLS_CODEC.get().decoder;
decoder.setByteBuffer(encoded.nioBuffer());
+ decoder.setMaxDecodeDepth(connection.getMaxDecodeDepth());
+
decoder.setZeroWidthArrayElementLimit(connection.getZeroWidthArrayElementLimit());
+
Section result = (Section) decoder.readObject();
decoder.setByteBuffer(null);
encoded.resetReaderIndex();
@@ -285,6 +291,8 @@ public final class AmqpCodec {
DecoderImpl decoder = getDecoder();
decoder.setBuffer(messageBytes);
+ decoder.setMaxDecodeDepth(consumer.getMaxDecodeDepth());
+
decoder.setZeroWidthArrayElementLimit(consumer.getZeroWidthArrayElementLimit());
Header header = null;
DeliveryAnnotations deliveryAnnotations = null;
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpTypedObjectDelegate.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpTypedObjectDelegate.java
index 88b7b9dd..88fee9fb 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpTypedObjectDelegate.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpTypedObjectDelegate.java
@@ -60,7 +60,7 @@ public class AmqpTypedObjectDelegate implements
AmqpObjectTypeDelegate {
Section body = null;
if (encodedBody != null) {
- body = AmqpCodec.decode(encodedBody);
+ body = AmqpCodec.decode(parent.getConnection(), encodedBody);
}
if (body == null) {
@@ -93,7 +93,7 @@ public class AmqpTypedObjectDelegate implements
AmqpObjectTypeDelegate {
// and decoding the value. Save the bytes for subsequent getObject
and
// copyInto calls to use.
encodedBody = AmqpCodec.encode(new AmqpValue(value));
- Section decodedBody = AmqpCodec.decode(encodedBody);
+ Section decodedBody = AmqpCodec.decode(parent.getConnection(),
encodedBody);
// This step requires a heavy-weight operation of both encoding
and decoding the
// incoming body value in order to create a copy such that changes
to the original
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java
index cc5573be..7b54c9b8 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java
@@ -407,22 +407,28 @@ public class ConnectionIntegrationTest extends
QpidJmsTestCase {
@Timeout(20)
public void testMaxFrameSizeOptionCommunicatedInOpen() throws Exception {
int frameSize = 39215;
- doMaxFrameSizeOptionTestImpl(frameSize,
UnsignedInteger.valueOf(frameSize));
+ doMaxFrameSizeOptionTestImpl("&amqp.maxFrameSize=" + frameSize,
UnsignedInteger.valueOf(frameSize));
}
@Test
@Timeout(20)
- public void testMaxFrameSizeOptionCommunicatedInOpenDefault() throws
Exception {
- doMaxFrameSizeOptionTestImpl(-1, UnsignedInteger.MAX_VALUE);
+ public void testMaxFrameSizeOptionCommunicatedInOpenProtonDefault() throws
Exception {
+ doMaxFrameSizeOptionTestImpl("&amqp.maxFrameSize=-1",
UnsignedInteger.valueOf(65535));
}
- private void doMaxFrameSizeOptionTestImpl(int uriOption, UnsignedInteger
transmittedValue) throws JMSException, InterruptedException, Exception,
IOException {
+ @Test
+ @Timeout(20)
+ public void testMaxFrameSizeCommunicatedInOpenDefault() throws Exception {
+ doMaxFrameSizeOptionTestImpl("", UnsignedInteger.valueOf(1 * 1024 *
1024));
+ }
+
+ private void doMaxFrameSizeOptionTestImpl(String uriOption,
UnsignedInteger transmittedValue) throws JMSException, InterruptedException,
Exception, IOException {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
testPeer.expectSaslLayerDisabledConnect(equalTo(transmittedValue));
// Each connection creates a session for managing temporary
destinations etc
testPeer.expectBegin();
- String uri = "amqp://localhost:" + testPeer.getServerPort() +
"?amqp.saslLayer=false&amqp.maxFrameSize=" + uriOption;
+ String uri = "amqp://localhost:" + testPeer.getServerPort() +
"?amqp.saslLayer=false" + uriOption;
ConnectionFactory factory = new JmsConnectionFactory(uri);
Connection connection = factory.createConnection();
connection.start();
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
index 9f91191b..65940563 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
@@ -20,6 +20,7 @@ package org.apache.qpid.jms.integration;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -28,8 +29,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -44,6 +47,7 @@ import jakarta.jms.Message;
import jakarta.jms.MessageConsumer;
import jakarta.jms.MessageListener;
import jakarta.jms.MessageProducer;
+import jakarta.jms.ObjectMessage;
import jakarta.jms.Queue;
import jakarta.jms.Session;
import jakarta.jms.TextMessage;
@@ -55,6 +59,7 @@ import org.apache.qpid.jms.JmsDefaultConnectionListener;
import org.apache.qpid.jms.JmsOperationTimedOutException;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.apache.qpid.jms.policy.JmsDefaultPrefetchPolicy;
+import org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport;
import org.apache.qpid.jms.test.QpidJmsTestCase;
import org.apache.qpid.jms.test.Wait;
import org.apache.qpid.jms.test.testpeer.AmqpPeerRunnable;
@@ -63,6 +68,7 @@ import org.apache.qpid.jms.test.testpeer.basictypes.AmqpError;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.AmqpValueDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.DataDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.HeaderDescribedType;
+import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.MessageAnnotationsDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.PropertiesDescribedType;
import org.apache.qpid.jms.test.testpeer.matchers.AcceptedMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.ModifiedMatcher;
@@ -2563,4 +2569,183 @@ public class ConsumerIntegrationTest extends
QpidJmsTestCase {
testPeer.waitForAllHandlersToComplete(2000);
}
}
+
+ @Test
+ @Timeout(20)
+ public void testMessageDecodeDepthLimit() throws Exception {
+ // check it allows without option
+ doReceiveMessageWithDepthLimitTestImpl(32, null, false);
+ // check it blocks with option set
+ doReceiveMessageWithDepthLimitTestImpl(32, 10, true);
+
+ // check it blocks without option
+ doReceiveMessageWithDepthLimitTestImpl(33, null, true);
+ // check it allows with option set
+ doReceiveMessageWithDepthLimitTestImpl(33, 33, false);
+ }
+
+ private void doReceiveMessageWithDepthLimitTestImpl(final int depth,
Integer depthLimitOption, boolean expectDeliveryFailure) throws Exception {
+ String optionsString = "?jms.receiveLocalOnly=true";
+ if(depthLimitOption != null) {
+ optionsString += "&amqp.maxDecodeDepth=" + depthLimitOption;
+ }
+
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ Connection connection = testFixture.establishConnecton(testPeer,
optionsString);
+ connection.start();
+
+ testPeer.expectBegin();
+
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ MessageAnnotationsDescribedType msgAnnotations = new
MessageAnnotationsDescribedType();
+
msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(),
AmqpMessageSupport.JMS_OBJECT_MESSAGE);
+
+ List<Object> lists = prepareNestedLists(depth);
+ DescribedType amqpValueContent = new AmqpValueDescribedType(lists);
+
+ testPeer.expectReceiverAttach();
+ testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations,
null, null, amqpValueContent);
+ if (expectDeliveryFailure) {
+ testPeer.expectDisposition(true, new
ModifiedMatcher().withDeliveryFailed(equalTo(true)).withUndeliverableHere(equalTo(true)));
+ } else {
+ testPeer.expectDispositionThatIsAcceptedAndSettled();
+ }
+ testPeer.expectClose();
+
+ MessageConsumer messageConsumer = session.createConsumer(queue);
+
+ Message receivedMessage;
+ if (expectDeliveryFailure) {
+ receivedMessage = messageConsumer.receive(250);
+
+ assertNull(receivedMessage);
+ } else {
+ receivedMessage = messageConsumer.receive(3000);
+
+ assertNotNull(receivedMessage);
+ assertTrue(receivedMessage instanceof ObjectMessage, "Expected
ObjectMessage instance, but got: " + receivedMessage.getClass().getName());
+ ObjectMessage objectMessage = (ObjectMessage)receivedMessage;
+
+ Object object = objectMessage.getObject();
+ assertNotNull(object, "Expected object but got null");
+ assertEquals(lists, object, "Message body object was not as
expected");
+ }
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(3000);
+ }
+ }
+
+ private static List<Object> prepareNestedLists(final int depth) {
+ final List<Object> body = new ArrayList<>();
+
+ List<Object> current = body;
+ for (int i = 1; i < depth; ++i) {
+ final List<Object> next = new ArrayList<>();
+
+ current.add(next);
+ current = next;
+ }
+
+ return body;
+ }
+
+ @Test
+ @Timeout(20)
+ public void testMessageZeroWidthArrayElementLimit() throws Exception {
+ // check it allows empty without option
+ doReceiveMessageWithZeroWidthArrayElementLimitTestImpl(0, null, false);
+ // check it blocks non-empty without option
+ doReceiveMessageWithZeroWidthArrayElementLimitTestImpl(10, null, true);
+ // check it blocks non-empty with option set lower
+ doReceiveMessageWithZeroWidthArrayElementLimitTestImpl(10, 9, true);
+ // check it allows non-empty with option set
+ doReceiveMessageWithZeroWidthArrayElementLimitTestImpl(10, 10, false);
+ }
+
+ private void doReceiveMessageWithZeroWidthArrayElementLimitTestImpl(final
int elements, Integer elementLmitOption, boolean expectDeliveryFailure) throws
Exception {
+ String optionsString = "?jms.receiveLocalOnly=true";
+ if (elementLmitOption != null) {
+ optionsString += "&amqp.zeroWidthArrayElementLimit=" +
elementLmitOption;
+ }
+
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ Connection connection = testFixture.establishConnecton(testPeer,
optionsString);
+ connection.start();
+
+ testPeer.expectBegin();
+
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ MessageAnnotationsDescribedType msgAnnotations = new
MessageAnnotationsDescribedType();
+
msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(),
AmqpMessageSupport.JMS_OBJECT_MESSAGE);
+
+ Binary payload =
prepareMessageEncodingWithArrayOfZeroWidthElements(elements);
+
+ testPeer.expectReceiverAttach();
+ // Expect that once receive is called, it flows credit.
+ testPeer.expectLinkFlow(false,
equalTo(UnsignedInteger.valueOf(1000)));
+ // Give it an initial (ie. more=true) transfer frame with header
only.
+ testPeer.sendTransferToLastOpenedLinkOnLastOpenedSession(payload,
1, "delivery1", false, 0);
+
+ if (expectDeliveryFailure) {
+ testPeer.expectDisposition(true, new
ModifiedMatcher().withDeliveryFailed(equalTo(true)).withUndeliverableHere(equalTo(true)));
+ } else {
+ testPeer.expectDispositionThatIsAcceptedAndSettled();
+ }
+ testPeer.expectClose();
+
+ MessageConsumer messageConsumer = session.createConsumer(queue);
+
+ Message receivedMessage;
+ if (expectDeliveryFailure) {
+ receivedMessage = messageConsumer.receive(250);
+
+ assertNull(receivedMessage);
+ } else {
+ receivedMessage = messageConsumer.receive(3000);
+
+ assertNotNull(receivedMessage);
+ assertTrue(receivedMessage instanceof ObjectMessage, "Expected
ObjectMessage instance, but got: " + receivedMessage.getClass().getName());
+ ObjectMessage objectMessage = (ObjectMessage)receivedMessage;
+
+ Object object = objectMessage.getObject();
+ assertNotNull(object, "Expected object but got null");
+
+ assertTrue(object.getClass().isArray());
+ assertEquals(boolean.class,
object.getClass().getComponentType());
+ final boolean[] expected = new boolean[elements];
+ Arrays.fill(expected, true);
+
+ assertArrayEquals(expected, (boolean[]) object, "Message body
object was not as expected");
+ }
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(3000);
+ }
+ }
+
+ private Binary prepareMessageEncodingWithArrayOfZeroWidthElements(int
elementCount) {
+ assertTrue(elementCount >= 0);
+ assertTrue(elementCount <= 255);
+
+ ByteBuffer buf = ByteBuffer.allocate(7);
+ buf.put((byte) 0x00); // DescribedType
+ buf.put((byte) 0x53); // small-ulong constructor
+ buf.put((byte) 0x77); // amqp-value ulong descriptor
+
+ buf.put((byte) 0xE0); // 'array8' type descriptor code
+ buf.put((byte) (1 + 1)); // 1 byte count + 1 byte type
+ buf.put((byte) elementCount); // count
+ buf.put((byte) 0x41); // boolean-true
+
+ Binary payload = new Binary(buf.array());
+
+ return payload;
+ }
}
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MultiTransferFrameMessageIntegrationTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MultiTransferFrameMessageIntegrationTest.java
index 7d6b2156..4aa3a63a 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MultiTransferFrameMessageIntegrationTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MultiTransferFrameMessageIntegrationTest.java
@@ -18,6 +18,9 @@
*/
package org.apache.qpid.jms.integration;
+import static
org.apache.qpid.proton.amqp.transport.LinkError.TRANSFER_LIMIT_EXCEEDED;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.fail;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -34,12 +37,14 @@ import jakarta.jms.MessageConsumer;
import jakarta.jms.Queue;
import jakarta.jms.Session;
+import org.apache.qpid.jms.exceptions.JmsConnectionFailedException;
import org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport;
import org.apache.qpid.jms.test.QpidJmsTestCase;
import org.apache.qpid.jms.test.testpeer.TestAmqpPeer;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.DataDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.MessageAnnotationsDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.PropertiesDescribedType;
+import org.apache.qpid.jms.test.testpeer.matchers.ErrorMatcher;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.DescribedType;
import org.apache.qpid.proton.amqp.UnsignedInteger;
@@ -127,4 +132,60 @@ public class MultiTransferFrameMessageIntegrationTest
extends QpidJmsTestCase {
return payload;
}
+
+ @Test
+ @Timeout(20)
+ public void testExceedsMaxTransfersPerDelivery() throws Exception {
+ doExceedsMaxTransfersPerDeliveryTestImpl(100_000, (11 * 100_000) -
1000, false, 10);
+ doExceedsMaxTransfersPerDeliveryTestImpl(100_000, (10 * 100_000) -
1000, true, 10);
+ }
+
+ private void doExceedsMaxTransfersPerDeliveryTestImpl(int
msgPayloadPerFrame, int payloadSizeInBytes, boolean sendLastEmptyTransfer, int
option) throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ Connection connection = testFixture.establishConnecton(testPeer,
"?jms.prefetchPolicy.all=0&amqp.maxTransfersPerDelivery=" + option);
+ connection.start();
+
+ testPeer.expectBegin();
+
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ PropertiesDescribedType properties = new PropertiesDescribedType();
+
properties.setContentType(AmqpMessageSupport.OCTET_STREAM_CONTENT_TYPE);
+
+ MessageAnnotationsDescribedType msgAnnotations = new
MessageAnnotationsDescribedType();
+
msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(),
AmqpMessageSupport.JMS_BYTES_MESSAGE);
+
+ final byte[] expectedContent =
createMessageBodyContent(payloadSizeInBytes, 0);
+ DescribedType dataContent = new DataDescribedType(new
Binary(expectedContent));
+
+ testPeer.expectReceiverAttach();
+
+ testPeer.expectLinkFlowAndSendBackMessages(null, msgAnnotations,
properties, null, dataContent, 1,
+ true, false,
Matchers.equalTo(UnsignedInteger.valueOf(1)), 1,
+ false, false,
msgPayloadPerFrame, sendLastEmptyTransfer);
+
+ ErrorMatcher errorMatcher = new ErrorMatcher()
+ .withCondition(equalTo(TRANSFER_LIMIT_EXCEEDED))
+ .withDescription(equalTo("Max transfers per delivery limit
exceeded"));
+
+ testPeer.expectClose(errorMatcher, false);
+
+ MessageConsumer messageConsumer = session.createConsumer(queue);
+
+ try {
+ messageConsumer.receiveNoWait();
+ fail("Expected exception");
+ } catch (JmsConnectionFailedException e) {
+ final String message = e.getMessage();
+ assertNotNull(message);
+ assertTrue(message.contains("Max transfers per delivery limit
exceeded"));
+ assertTrue(message.contains("transfer-limit-exceeded"));
+ }
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(3000);
+ }
+ }
}
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
index b5036dc3..87d42730 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
@@ -31,7 +31,9 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -43,6 +45,7 @@ import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.AmqpValueDescri
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.DataDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.MessageAnnotationsDescribedType;
import
org.apache.qpid.jms.test.testpeer.describedtypes.sections.PropertiesDescribedType;
+import org.apache.qpid.jms.test.testpeer.matchers.ModifiedMatcher;
import
org.apache.qpid.jms.test.testpeer.matchers.sections.MessageAnnotationsSectionMatcher;
import
org.apache.qpid.jms.test.testpeer.matchers.sections.MessageHeaderSectionMatcher;
import
org.apache.qpid.jms.test.testpeer.matchers.sections.MessagePropertiesSectionMatcher;
@@ -52,6 +55,7 @@ import
org.apache.qpid.jms.test.testpeer.matchers.types.EncodedDataMatcher;
import org.apache.qpid.jms.util.SimplePojo;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.DescribedType;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderFactoryTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderFactoryTest.java
index 1b40c26d..9791cf00 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderFactoryTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderFactoryTest.java
@@ -158,4 +158,40 @@ public class AmqpProviderFactoryTest extends
QpidJmsTestCase {
assertEquals("v+host", amqpProvider.getVhost());
}
+
+ @Test
+ @Timeout(20)
+ public void testCreateProviderAppliesMaxTransfersURIOption() throws
IOException, Exception {
+ int transfers = 42798;
+ Provider provider = AmqpProviderFactory.create(new
URI(peerURI.toString() + "?amqp.maxTransfersPerDelivery=" + transfers));
+ assertNotNull(provider);
+ assertTrue(provider instanceof AmqpProvider);
+ AmqpProvider amqpProvider = (AmqpProvider) provider;
+
+ assertEquals(transfers, amqpProvider.getMaxTransfersPerDelivery(),
"maxTransfersPerDelivery option was not applied");
+ }
+
+ @Test
+ @Timeout(20)
+ public void testCreateProviderAppliesMaxDecodeDepthURIOption() throws
IOException, Exception {
+ int depth = 17;
+ Provider provider = AmqpProviderFactory.create(new
URI(peerURI.toString() + "?amqp.maxDecodeDepth=" + depth));
+ assertNotNull(provider);
+ assertTrue(provider instanceof AmqpProvider);
+ AmqpProvider amqpProvider = (AmqpProvider) provider;
+
+ assertEquals(depth, amqpProvider.getMaxDecodeDepth(), "maxDecodeDepth
option was not applied");
+ }
+
+ @Test
+ @Timeout(20)
+ public void testCreateProviderAppliesZeroWidthArrayElementLimitURIOption()
throws IOException, Exception {
+ int elements = 24;
+ Provider provider = AmqpProviderFactory.create(new
URI(peerURI.toString() + "?amqp.zeroWidthArrayElementLimit=" + elements));
+ assertNotNull(provider);
+ assertTrue(provider instanceof AmqpProvider);
+ AmqpProvider amqpProvider = (AmqpProvider) provider;
+
+ assertEquals(elements, amqpProvider.getZeroWidthArrayElementLimit(),
"zeroWidthArrayElementLimit option was not applied");
+ }
}
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderTest.java
index d00bfa13..f2ecd20c 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/AmqpProviderTest.java
@@ -271,6 +271,58 @@ public class AmqpProviderTest extends QpidJmsTestCase {
}
}
+ @Test
+ @Timeout(20)
+ public void testSetMaxTransfersPerDelivery() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer()) {
+ testPeer.expectSaslAnonymous();
+
+ provider = new
AmqpProviderFactory().createProvider(getPeerURI(testPeer));
+
+ TransportImpl transport = (TransportImpl)
provider.getProtonTransport();
+
+ final int NEW_MAX = 500;
+
+ provider.setMaxTransfersPerDelivery(NEW_MAX);
+ provider.connect(connectionInfo);
+
+ assertEquals(NEW_MAX, transport.getMaxTransfersPerDelivery());
+
+ testPeer.expectOpen();
+ testPeer.expectClose();
+
+ provider.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
+ @Test
+ @Timeout(20)
+ public void testSkipSetMaxTransfersPerDelivery() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer()) {
+ testPeer.expectSaslAnonymous();
+
+ provider = new
AmqpProviderFactory().createProvider(getPeerURI(testPeer));
+
+ TransportImpl transport = (TransportImpl)
provider.getProtonTransport();
+
+ final int RECORDED_MAX = transport.getMaxTransfersPerDelivery();
+
+ provider.setMaxTransfersPerDelivery(-1);
+ provider.connect(connectionInfo);
+
+ assertEquals(RECORDED_MAX, transport.getMaxTransfersPerDelivery());
+
+ testPeer.expectOpen();
+ testPeer.expectClose();
+
+ provider.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
@Test
@Timeout(20)
public void testStartThrowsIfNoListenerSet() throws Exception {
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodecTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodecTest.java
index d86e7110..7da0d80b 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodecTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodecTest.java
@@ -21,17 +21,22 @@
package org.apache.qpid.jms.provider.amqp.message;
import static
org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport.encodeMessage;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import java.io.IOException;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -64,6 +69,8 @@ import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Header;
import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
+import org.apache.qpid.proton.codec.DecodeException;
+import org.apache.qpid.proton.codec.ReadableBuffer.ByteBufferReader;
import org.apache.qpid.proton.message.Message;
import org.apache.qpid.proton.message.impl.MessageImpl;
import org.junit.jupiter.api.BeforeEach;
@@ -75,6 +82,9 @@ import io.netty.buffer.ByteBuf;
public class AmqpCodecTest extends QpidJmsTestCase {
+ private static final int MAX_DECODE_DEPTH = 32;
+ private static final int ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT = 0;
+
private AmqpConsumer mockConsumer;
private AmqpConnection mockConnection;
@@ -85,8 +95,13 @@ public class AmqpCodecTest extends QpidJmsTestCase {
JmsConsumerId consumerId = new JmsConsumerId("ID:MOCK:1", 1, 1);
mockConnection = Mockito.mock(AmqpConnection.class);
+
Mockito.when(mockConnection.getMaxDecodeDepth()).thenReturn(MAX_DECODE_DEPTH);
+
Mockito.when(mockConnection.getZeroWidthArrayElementLimit()).thenReturn(ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT);
mockConsumer = Mockito.mock(AmqpConsumer.class);
Mockito.when(mockConsumer.getResourceInfo()).thenReturn(new
JmsConsumerInfo(consumerId, null));
+ Mockito.when(mockConsumer.getConnection()).thenReturn(mockConnection);
+
Mockito.when(mockConsumer.getMaxDecodeDepth()).thenReturn(MAX_DECODE_DEPTH);
+
Mockito.when(mockConsumer.getZeroWidthArrayElementLimit()).thenReturn(ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT);
}
//----- AmqpHeader encode and decode
-------------------------------------//
@@ -96,7 +111,7 @@ public class AmqpCodecTest extends QpidJmsTestCase {
Header empty = new Header();
ByteBuf encoded = AmqpCodec.encode(empty);
- Header decoded = (Header) AmqpCodec.decode(encoded);
+ Header decoded = (Header) AmqpCodec.decode(mockConnection, encoded);
assertNotNull(decoded);
@@ -115,7 +130,7 @@ public class AmqpCodecTest extends QpidJmsTestCase {
header.setDurable(true);
ByteBuf encoded = AmqpCodec.encode(header.getHeader());
- AmqpHeader decoded = new AmqpHeader((Header)
AmqpCodec.decode(encoded));
+ AmqpHeader decoded = new AmqpHeader((Header)
AmqpCodec.decode(mockConnection, encoded));
assertTrue(decoded.isDurable());
assertEquals(4, decoded.getPriority());
@@ -130,7 +145,7 @@ public class AmqpCodecTest extends QpidJmsTestCase {
header.setDeliveryCount(1);
ByteBuf encoded = AmqpCodec.encode(header.getHeader());
- AmqpHeader decoded = new AmqpHeader((Header)
AmqpCodec.decode(encoded));
+ AmqpHeader decoded = new AmqpHeader((Header)
AmqpCodec.decode(mockConnection, encoded));
assertFalse(decoded.isDurable());
assertEquals(4, decoded.getPriority());
@@ -149,7 +164,7 @@ public class AmqpCodecTest extends QpidJmsTestCase {
header.setTimeToLive(32768);
ByteBuf encoded = AmqpCodec.encode(header.getHeader());
- AmqpHeader decoded = new AmqpHeader((Header)
AmqpCodec.decode(encoded));
+ AmqpHeader decoded = new AmqpHeader((Header)
AmqpCodec.decode(mockConnection, encoded));
assertTrue(decoded.isDurable());
assertTrue(decoded.isFirstAcquirer());
@@ -879,6 +894,100 @@ public class AmqpCodecTest extends QpidJmsTestCase {
assertTrue(delegate instanceof AmqpTypedObjectDelegate, "Unexpected
delegate type: " + delegate);
}
+ @Test
+ public void testDecodeDepthLimit() throws Exception {
+ Message message = Proton.message();
+
+ // Verify it fails if exceeding the limit.
+ List<Object> lists = prepareNestedLists(MAX_DECODE_DEPTH + 1);
+ message.setBody(new AmqpValue(lists));
+
+ try {
+ AmqpCodec.decodeMessage(mockConsumer, encodeMessage(message));
+ fail("Expected exception");
+ } catch(DecodeException de) {
+ //Expected
+ }
+
+ // Now check it works if not exceeding the limit.
+ lists = prepareNestedLists(MAX_DECODE_DEPTH);
+ message.setBody(new AmqpValue(lists));
+
+ JmsMessage jmsMessage = AmqpCodec.decodeMessage(mockConsumer,
encodeMessage(message)).asJmsMessage();
+ assertNotNull(jmsMessage, "Message should not be null");
+ assertEquals(JmsObjectMessage.class, jmsMessage.getClass(),
"Unexpected message class type");
+
+ JmsMessageFacade facade = jmsMessage.getFacade();
+ assertNotNull(facade, "Facade should not be null");
+ assertEquals(AmqpJmsObjectMessageFacade.class, facade.getClass(),
"Unexpected facade class type");
+
+ assertEquals(lists, ((AmqpJmsObjectMessageFacade) facade).getObject());
+ }
+
+ @Test
+ public void testZeroWidthArrayElementLimit() throws Exception {
+ // Verify it fails if exceeding the limit.
+ ByteBuffer buf =
prepareMessageEncodingWithZeroWidthArray(ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT + 1);
+
+ try {
+ AmqpCodec.decodeMessage(mockConsumer, ByteBufferReader.wrap(buf));
+ fail("Expected exception");
+ } catch(DecodeException de) {
+ //Expected
+ }
+
+ // Now check it works if not exceeding the limit.
+ assertEquals(0, ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT);
+ buf = prepareMessageEncodingWithZeroWidthArray(0);
+
+ doZeroWidthArrayElementLimitTestImpl(buf, new boolean[0]);
+
+ // Now check it works if setting a higher limit.
+ final int count = 5;
+ buf = prepareMessageEncodingWithZeroWidthArray(count);
+
+ final boolean[] expected = new boolean[count];
+ Arrays.fill(expected, true);
+
+
Mockito.when(mockConsumer.getZeroWidthArrayElementLimit()).thenReturn(count);
+
+ doZeroWidthArrayElementLimitTestImpl(buf, expected);
+ }
+
+ private void doZeroWidthArrayElementLimitTestImpl(ByteBuffer buf, final
boolean[] expected) throws Exception {
+ JmsMessage jmsMessage = AmqpCodec.decodeMessage(mockConsumer,
ByteBufferReader.wrap(buf)).asJmsMessage();
+ assertNotNull(jmsMessage, "Message should not be null");
+ assertEquals(JmsObjectMessage.class, jmsMessage.getClass(),
"Unexpected message class type");
+
+ JmsMessageFacade facade = jmsMessage.getFacade();
+ assertNotNull(facade, "Facade should not be null");
+ assertEquals(AmqpJmsObjectMessageFacade.class, facade.getClass(),
"Unexpected facade class type");
+
+ final Serializable object = ((AmqpJmsObjectMessageFacade)
facade).getObject();
+ assertTrue(object.getClass().isArray());
+ assertEquals(boolean.class, object.getClass().getComponentType());
+ assertArrayEquals(expected, (boolean[]) object);
+ }
+
+ private ByteBuffer prepareMessageEncodingWithZeroWidthArray(int
elementCount) {
+ assertTrue(elementCount >= 0);
+ assertTrue(elementCount <= 255);
+
+ ByteBuffer buf = ByteBuffer.allocate(7);
+ buf.put((byte) 0x00); // DescribedType
+ buf.put((byte) 0x53); // small-ulong constructor
+ buf.put((byte) 0x77); // amqp-value ulong descriptor
+
+ buf.put((byte) 0xE0); // 'array8' type descriptor code
+ buf.put((byte) (1 + 1)); // 1 byte count + 1 byte type
+ buf.put((byte) elementCount); // count
+ buf.put((byte) 0x41); // boolean-true
+
+ buf.flip();
+
+ return buf;
+ }
+
// --------- AmqpSequence Body Section ---------
/**
@@ -1033,4 +1142,18 @@ public class AmqpCodecTest extends QpidJmsTestCase {
return message;
}
+
+ private static List<Object> prepareNestedLists(final int depth) {
+ final List<Object> body = new ArrayList<>();
+
+ List<Object> current = body;
+ for (int i = 1; i < depth; ++i) {
+ final List<Object> next = new ArrayList<>();
+
+ current.add(next);
+ current = next;
+ }
+
+ return body;
+ }
}
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageTypesTestCase.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageTypesTestCase.java
index 3e2c1fc1..1ee03a4f 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageTypesTestCase.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageTypesTestCase.java
@@ -32,6 +32,9 @@ import org.mockito.Mockito;
public class AmqpJmsMessageTypesTestCase extends QpidJmsTestCase {
+ private static final int MAX_DECODE_DEPTH = 32;
+ private static final int ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT = 0;
+
private JmsDestination consumerDestination;
@BeforeEach
@@ -133,6 +136,9 @@ public class AmqpJmsMessageTypesTestCase extends
QpidJmsTestCase {
Mockito.when(consumer.getConnection()).thenReturn(connection);
Mockito.when(consumer.getDestination()).thenReturn(consumerDestination);
Mockito.when(consumer.getResourceInfo()).thenReturn(new
JmsConsumerInfo(consumerId, null));
+
Mockito.when(consumer.getMaxDecodeDepth()).thenReturn(MAX_DECODE_DEPTH);
+
Mockito.when(consumer.getZeroWidthArrayElementLimit()).thenReturn(ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT);
+
return consumer;
}
@@ -145,6 +151,8 @@ public class AmqpJmsMessageTypesTestCase extends
QpidJmsTestCase {
AmqpConnection connection = Mockito.mock(AmqpConnection.class);
Mockito.when(connection.getResourceInfo()).thenReturn(new
JmsConnectionInfo(connectionId));
Mockito.when(connection.isObjectMessageUsesAmqpTypes()).thenReturn(amqpTyped);
+
Mockito.when(connection.getMaxDecodeDepth()).thenReturn(MAX_DECODE_DEPTH);
+
Mockito.when(connection.getZeroWidthArrayElementLimit()).thenReturn(ZERO_WIDTH_ARRAY_ELEMENTS_LIMIT);
return connection;
}
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java
index 2906f74a..c5634177 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java
@@ -2783,6 +2783,16 @@ public class TestAmqpPeer implements AutoCloseable
final String
tagAsString,
final Boolean
more,
final int
sendDelay) {
+ Binary payload = prepareTransferPayload(headerDescribedType,
messageAnnotationsDescribedType, propertiesDescribedType,
appPropertiesDescribedType, content);
+
+ sendTransferToLastOpenedLinkOnLastOpenedSession(payload,
nextIncomingDeliveryId, tagAsString, more, sendDelay);
+ }
+
+ public void sendTransferToLastOpenedLinkOnLastOpenedSession(final Binary
payload,
+ final int
nextIncomingDeliveryId,
+ final String
tagAsString,
+ final Boolean
more,
+ final int
sendDelay) {
synchronized (_handlersLock) {
CompositeAmqpPeerRunnable comp =
insertCompsiteActionForLastHandler();
@@ -2801,8 +2811,6 @@ public class TestAmqpPeer implements AutoCloseable
transferResponse.setMore(more);
}
- Binary payload = prepareTransferPayload(headerDescribedType,
messageAnnotationsDescribedType, propertiesDescribedType,
appPropertiesDescribedType, content);
-
// The response frame channel will be dynamically set based on the
incoming frame. Using the -1 is an illegal placeholder.
final FrameSender transferSender = new FrameSender(this,
FrameType.AMQP, -1, transferResponse, payload);
transferSender.setValueProvider(new ValueProvider()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
