You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2016/12/13 17:37:17 UTC
activemq git commit: https://issues.apache.org/jira/browse/AMQ-6536
Repository: activemq
Updated Branches:
refs/heads/master 25703fbd1 -> 4a1c05b62
https://issues.apache.org/jira/browse/AMQ-6536
Add support for the receiver side of a sender link which carries a
desired capabilities array which can request to know if we support
delayed delivery, answer with an offered capability in that case.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/4a1c05b6
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/4a1c05b6
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/4a1c05b6
Branch: refs/heads/master
Commit: 4a1c05b628fbb3c97c394eebc6810a9786daca4e
Parents: 25703fb
Author: Timothy Bish <ta...@gmail.com>
Authored: Tue Dec 13 12:36:52 2016 -0500
Committer: Timothy Bish <ta...@gmail.com>
Committed: Tue Dec 13 12:36:59 2016 -0500
----------------------------------------------------------------------
.../transport/amqp/protocol/AmqpSession.java | 18 +++--
.../transport/amqp/client/AmqpSender.java | 38 +++++++++-
.../transport/amqp/client/AmqpSession.java | 76 +++++++++++++++++++-
.../transport/amqp/client/AmqpSupport.java | 1 +
.../amqp/client/util/UnmodifiableLink.java | 31 ++++++++
.../amqp/client/util/UnmodifiableReceiver.java | 6 ++
.../amqp/client/util/UnmodifiableSender.java | 6 ++
.../amqp/client/util/UnmodifiableSession.java | 47 ++++++++++++
.../transport/amqp/interop/AmqpSenderTest.java | 38 ++++++++++
pom.xml | 2 +-
10 files changed, 253 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSession.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSession.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSession.java
index 0527211..6a0cbf7 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSession.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSession.java
@@ -26,7 +26,9 @@ import static org.apache.activemq.transport.amqp.AmqpSupport.createDestination;
import static org.apache.activemq.transport.amqp.AmqpSupport.findFilter;
import java.io.IOException;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import javax.jms.InvalidSelectorException;
@@ -72,7 +74,7 @@ public class AmqpSession implements AmqpResource {
private static final Logger LOG = LoggerFactory.getLogger(AmqpSession.class);
- private final Map<ConsumerId, AmqpSender> consumers = new HashMap<ConsumerId, AmqpSender>();
+ private final Map<ConsumerId, AmqpSender> consumers = new HashMap<>();
private final AmqpConnection connection;
private final Session protonSession;
@@ -190,7 +192,7 @@ public class AmqpSession implements AmqpResource {
if (target.getDynamic()) {
destination = connection.createTemporaryDestination(protonReceiver, target.getCapabilities());
- Map<Symbol, Object> dynamicNodeProperties = new HashMap<Symbol, Object>();
+ Map<Symbol, Object> dynamicNodeProperties = new HashMap<>();
dynamicNodeProperties.put(LIFETIME_POLICY, DeleteOnClose.getInstance());
// Currently we only support temporary destinations with delete on close lifetime policy.
@@ -218,6 +220,14 @@ public class AmqpSession implements AmqpResource {
}
}
+ Symbol[] remoteDesiredCapabilities = protonReceiver.getRemoteDesiredCapabilities();
+ if (remoteDesiredCapabilities != null) {
+ List<Symbol> list = Arrays.asList(remoteDesiredCapabilities);
+ if (list.contains(AmqpSupport.DELAYED_DELIVERY)) {
+ protonReceiver.setOfferedCapabilities(new Symbol[] { AmqpSupport.DELAYED_DELIVERY });
+ }
+ }
+
receiver.setDestination(destination);
connection.sendToActiveMQ(producerInfo, new ResponseHandler() {
@Override
@@ -255,7 +265,7 @@ public class AmqpSession implements AmqpResource {
LOG.debug("opening new sender {} on link: {}", consumerInfo.getConsumerId(), protonSender.getName());
try {
- final Map<Symbol, Object> supportedFilters = new HashMap<Symbol, Object>();
+ final Map<Symbol, Object> supportedFilters = new HashMap<>();
protonSender.setContext(sender);
boolean noLocal = false;
@@ -311,7 +321,7 @@ public class AmqpSession implements AmqpResource {
} else if (source.getDynamic()) {
destination = connection.createTemporaryDestination(protonSender, source.getCapabilities());
- Map<Symbol, Object> dynamicNodeProperties = new HashMap<Symbol, Object>();
+ Map<Symbol, Object> dynamicNodeProperties = new HashMap<>();
dynamicNodeProperties.put(LIFETIME_POLICY, DeleteOnClose.getInstance());
// Currently we only support temporary destinations with delete on close lifetime policy.
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSender.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSender.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSender.java
index 3850ec7..8d28688 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSender.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSender.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -70,9 +71,14 @@ public class AmqpSender extends AmqpAbstractResource<Sender> {
private boolean presettle;
private long sendTimeout = DEFAULT_SEND_TIMEOUT;
- private final Set<Delivery> pending = new LinkedHashSet<Delivery>();
+ private final Set<Delivery> pending = new LinkedHashSet<>();
private byte[] encodeBuffer = new byte[1024 * 8];
+ private Symbol[] desiredCapabilities;
+ private Symbol[] offeredCapabilities;
+ private Map<Symbol, Object> properties;
+
+
/**
* Create a new sender instance.
*
@@ -245,6 +251,30 @@ public class AmqpSender extends AmqpAbstractResource<Sender> {
this.sendTimeout = sendTimeout;
}
+ public void setDesiredCapabilities(Symbol[] desiredCapabilities) {
+ if (getEndpoint() != null) {
+ throw new IllegalStateException("Endpoint already established");
+ }
+
+ this.desiredCapabilities = desiredCapabilities;
+ }
+
+ public void setOfferedCapabilities(Symbol[] offeredCapabilities) {
+ if (getEndpoint() != null) {
+ throw new IllegalStateException("Endpoint already established");
+ }
+
+ this.offeredCapabilities = offeredCapabilities;
+ }
+
+ public void setProperties(Map<Symbol, Object> properties) {
+ if (getEndpoint() != null) {
+ throw new IllegalStateException("Endpoint already established");
+ }
+
+ this.properties = properties;
+ }
+
//----- Private Sender implementation ------------------------------------//
private void checkClosed() {
@@ -279,6 +309,10 @@ public class AmqpSender extends AmqpAbstractResource<Sender> {
}
sender.setReceiverSettleMode(ReceiverSettleMode.FIRST);
+ sender.setDesiredCapabilities(desiredCapabilities);
+ sender.setOfferedCapabilities(offeredCapabilities);
+ sender.setProperties(properties);
+
setEndpoint(sender);
super.doOpen();
@@ -408,7 +442,7 @@ public class AmqpSender extends AmqpAbstractResource<Sender> {
@Override
public void processDeliveryUpdates(AmqpConnection connection) throws IOException {
- List<Delivery> toRemove = new ArrayList<Delivery>();
+ List<Delivery> toRemove = new ArrayList<>();
for (Delivery delivery : pending) {
DeliveryState state = delivery.getRemoteState();
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
index b239dae..88cba94 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
@@ -17,6 +17,7 @@
package org.apache.activemq.transport.amqp.client;
import java.io.IOException;
+import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -24,6 +25,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.transport.amqp.client.util.AsyncResult;
import org.apache.activemq.transport.amqp.client.util.ClientFuture;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableSession;
+import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Source;
import org.apache.qpid.proton.amqp.messaging.Target;
import org.apache.qpid.proton.engine.Connection;
@@ -87,7 +89,7 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
* @throws Exception if an error occurs while creating the sender.
*/
public AmqpSender createSender() throws Exception {
- return createSender(null, false);
+ return createSender(null, false, null, null, null);
}
/**
@@ -101,14 +103,30 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
* @throws Exception if an error occurs while creating the sender.
*/
public AmqpSender createSender(final String address) throws Exception {
- return createSender(address, false);
+ return createSender(address, false, null, null, null);
}
/**
* Create a sender instance using the given address
*
* @param address
- * the address to which the sender will produce its messages.
+ * the address to which the sender will produce its messages.
+ * @param desiredCapabilities
+ * the capabilities that the caller wants the remote to support.
+ *
+ * @return a newly created sender that is ready for use.
+ *
+ * @throws Exception if an error occurs while creating the sender.
+ */
+ public AmqpSender createSender(final String address, Symbol[] desiredCapabilities) throws Exception {
+ return createSender(address, false, desiredCapabilities, null, null);
+ }
+
+ /**
+ * Create a sender instance using the given address
+ *
+ * @param address
+ * the address to which the sender will produce its messages.
* @param presettle
* controls if the created sender produces message that have already been marked settled.
*
@@ -117,10 +135,36 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
* @throws Exception if an error occurs while creating the sender.
*/
public AmqpSender createSender(final String address, boolean presettle) throws Exception {
+ return createSender(address, presettle, null, null, null);
+ }
+
+ /**
+ * Create a sender instance using the given address
+ *
+ * @param address
+ * the address to which the sender will produce its messages.
+ * @param presettle
+ * controls if the created sender produces message that have already been marked settled.
+ * @param desiredCapabilities
+ * the capabilities that the caller wants the remote to support.
+ * @param offeredCapabilities
+ * the capabilities that the caller wants the advertise support for.
+ * @param properties
+ * the properties to send as part of the sender open.
+ *
+ * @return a newly created sender that is ready for use.
+ *
+ * @throws Exception if an error occurs while creating the sender.
+ */
+ public AmqpSender createSender(final String address, boolean presettle, Symbol[] desiredCapabilities, Symbol[] offeredCapabilities, Map<Symbol, Object> properties) throws Exception {
checkClosed();
final AmqpSender sender = new AmqpSender(AmqpSession.this, address, getNextSenderId());
sender.setPresettle(presettle);
+ sender.setDesiredCapabilities(desiredCapabilities);
+ sender.setOfferedCapabilities(offeredCapabilities);
+ sender.setProperties(properties);
+
final ClientFuture request = new ClientFuture();
connection.getScheduler().execute(new Runnable() {
@@ -166,9 +210,35 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
* @throws Exception if an error occurs while creating the receiver.
*/
public AmqpSender createSender(Target target, String senderId) throws Exception {
+ return createSender(target, senderId, null, null, null);
+ }
+
+ /**
+ * Create a sender instance using the given Target
+ *
+ * @param target
+ * the caller created and configured Target used to create the sender link.
+ * @param sender
+ * the sender ID to assign to the newly created Sender.
+ * @param desiredCapabilities
+ * the capabilities that the caller wants the remote to support.
+ * @param offeredCapabilities
+ * the capabilities that the caller wants the advertise support for.
+ * @param properties
+ * the properties to send as part of the sender open.
+ *
+ * @return a newly created sender that is ready for use.
+ *
+ * @throws Exception if an error occurs while creating the receiver.
+ */
+ public AmqpSender createSender(Target target, String senderId, Symbol[] desiredCapabilities, Symbol[] offeredCapabilities, Map<Symbol, Object> properties) throws Exception {
checkClosed();
final AmqpSender sender = new AmqpSender(AmqpSession.this, target, senderId);
+ sender.setDesiredCapabilities(desiredCapabilities);
+ sender.setOfferedCapabilities(offeredCapabilities);
+ sender.setProperties(properties);
+
final ClientFuture request = new ClientFuture();
connection.getScheduler().execute(new Runnable() {
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSupport.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSupport.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSupport.java
index 9aca5b8..8ffcfbd 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSupport.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSupport.java
@@ -39,6 +39,7 @@ public class AmqpSupport {
// Symbols used for connection capabilities
public static final Symbol SOLE_CONNECTION_CAPABILITY = Symbol.valueOf("sole-connection-for-container");
public static final Symbol ANONYMOUS_RELAY = Symbol.valueOf("ANONYMOUS-RELAY");
+ public static final Symbol DELAYED_DELIVERY = Symbol.valueOf("DELAYED_DELIVERY");
// Symbols used to announce connection error information
public static final Symbol CONNECTION_OPEN_FAILED = Symbol.valueOf("amqp:connection-establishment-failed");
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
index 2f7ff8e..9b17d86 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
@@ -254,6 +254,7 @@ public class UnmodifiableLink implements Link {
return link.detached();
}
+ @Override
public Record attachments() {
return link.attachments();
}
@@ -272,4 +273,34 @@ public class UnmodifiableLink implements Link {
public Map<Symbol, Object> getRemoteProperties() {
return link.getRemoteProperties();
}
+
+ @Override
+ public Symbol[] getDesiredCapabilities() {
+ return link.getDesiredCapabilities();
+ }
+
+ @Override
+ public Symbol[] getOfferedCapabilities() {
+ return link.getOfferedCapabilities();
+ }
+
+ @Override
+ public Symbol[] getRemoteDesiredCapabilities() {
+ return link.getRemoteDesiredCapabilities();
+ }
+
+ @Override
+ public Symbol[] getRemoteOfferedCapabilities() {
+ return link.getRemoteOfferedCapabilities();
+ }
+
+ @Override
+ public void setDesiredCapabilities(Symbol[] capabilities) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
+
+ @Override
+ public void setOfferedCapabilities(Symbol[] capabilities) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableReceiver.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableReceiver.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableReceiver.java
index 1b07ed0..121a7f5 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableReceiver.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableReceiver.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.transport.amqp.client.util;
+import org.apache.qpid.proton.codec.WritableBuffer;
import org.apache.qpid.proton.engine.Receiver;
/**
@@ -56,4 +57,9 @@ public class UnmodifiableReceiver extends UnmodifiableLink implements Receiver {
public void setDrain(boolean drain) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
+
+ @Override
+ public int recv(WritableBuffer buffer) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSender.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSender.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSender.java
index 1517a93..fbe5772 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSender.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSender.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.transport.amqp.client.util;
+import org.apache.qpid.proton.codec.ReadableBuffer;
import org.apache.qpid.proton.engine.Sender;
/**
@@ -42,4 +43,9 @@ public class UnmodifiableSender extends UnmodifiableLink implements Sender {
public void abort() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
+
+ @Override
+ public int send(ReadableBuffer buffer) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSession.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSession.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSession.java
index 31b0452..055bb51 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSession.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSession.java
@@ -17,7 +17,9 @@
package org.apache.activemq.transport.amqp.client.util;
import java.util.EnumSet;
+import java.util.Map;
+import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EndpointState;
@@ -147,4 +149,49 @@ public class UnmodifiableSession implements Session {
public void setOutgoingWindow(long outgoingWindowSize) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
+
+ @Override
+ public Symbol[] getDesiredCapabilities() {
+ return session.getDesiredCapabilities();
+ }
+
+ @Override
+ public Symbol[] getOfferedCapabilities() {
+ return session.getOfferedCapabilities();
+ }
+
+ @Override
+ public Map<Symbol, Object> getProperties() {
+ return session.getProperties();
+ }
+
+ @Override
+ public Symbol[] getRemoteDesiredCapabilities() {
+ return session.getRemoteDesiredCapabilities();
+ }
+
+ @Override
+ public Symbol[] getRemoteOfferedCapabilities() {
+ return session.getRemoteOfferedCapabilities();
+ }
+
+ @Override
+ public Map<Symbol, Object> getRemoteProperties() {
+ return session.getRemoteProperties();
+ }
+
+ @Override
+ public void setDesiredCapabilities(Symbol[] capabilities) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
+
+ @Override
+ public void setOfferedCapabilities(Symbol[] capabilities) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
+
+ @Override
+ public void setProperties(Map<Symbol, Object> capabilities) {
+ throw new UnsupportedOperationException("Cannot alter the Link state");
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSenderTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSenderTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSenderTest.java
index 3c3b75d..da9e011 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSenderTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSenderTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.transport.amqp.interop;
+import static org.apache.activemq.transport.amqp.AmqpSupport.contains;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -23,6 +24,7 @@ import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.QueueViewMBean;
import org.apache.activemq.broker.jmx.TopicViewMBean;
import org.apache.activemq.transport.amqp.client.AmqpClient;
@@ -31,8 +33,10 @@ import org.apache.activemq.transport.amqp.client.AmqpConnection;
import org.apache.activemq.transport.amqp.client.AmqpMessage;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.activemq.transport.amqp.client.AmqpSupport;
import org.apache.activemq.transport.amqp.client.AmqpValidator;
import org.apache.activemq.util.Wait;
+import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Sender;
import org.junit.Test;
@@ -206,4 +210,38 @@ public class AmqpSenderTest extends AmqpClientTestSupport {
sender.close();
connection.close();
}
+
+ @Test
+ public void testDeliveryDelayOfferedWhenRequested() throws Exception {
+
+ final BrokerViewMBean brokerView = getProxyToBroker();
+
+ AmqpClient client = createAmqpClient();
+ client.setValidator(new AmqpValidator() {
+
+ @Override
+ public void inspectOpenedResource(Sender sender) {
+
+ Symbol[] offered = sender.getRemoteOfferedCapabilities();
+ if (!contains(offered, AmqpSupport.DELAYED_DELIVERY)) {
+ markAsInvalid("Broker did not indicate it support delayed message delivery");
+ }
+ }
+ });
+
+ AmqpConnection connection = trackConnection(client.connect());
+ AmqpSession session = connection.createSession();
+
+ assertEquals(0, brokerView.getQueues().length);
+
+ AmqpSender sender = session.createSender("queue://" + getTestName(), new Symbol[] { AmqpSupport.DELAYED_DELIVERY });
+ assertNotNull(sender);
+
+ assertEquals(1, brokerView.getQueues().length);
+
+ connection.getStateInspector().assertValid();
+
+ sender.close();
+ connection.close();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a1c05b6/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index c4479c0..37c913c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -103,7 +103,7 @@
<p2psockets-version>1.1.2</p2psockets-version>
<linkedin-zookeeper-version>1.4.0</linkedin-zookeeper-version>
<zookeeper-version>3.4.6</zookeeper-version>
- <qpid-proton-version>0.15.0</qpid-proton-version>
+ <qpid-proton-version>0.16.0</qpid-proton-version>
<qpid-jms-version>0.11.1</qpid-jms-version>
<qpid-jms-netty-version>4.0.41.Final</qpid-jms-netty-version>
<qpid-jms-proton-version>0.14.0</qpid-jms-proton-version>