You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2017/04/28 17:31:06 UTC
qpid-jms git commit: QPIDJMS-289: set JMSDeliveryTime during send
even if there is no delivery delay,
and use timestamp on receive if the annotation isnt present
Repository: qpid-jms
Updated Branches:
refs/heads/master fafc3d3c6 -> 70fe1b882
QPIDJMS-289: set JMSDeliveryTime during send even if there is no delivery delay, and use timestamp on receive if the annotation isnt present
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/70fe1b88
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/70fe1b88
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/70fe1b88
Branch: refs/heads/master
Commit: 70fe1b882e1381b1016a10b0707f6e112d4e0598
Parents: fafc3d3
Author: Robert Gemmell <ro...@apache.org>
Authored: Fri Apr 28 18:28:18 2017 +0100
Committer: Robert Gemmell <ro...@apache.org>
Committed: Fri Apr 28 18:28:18 2017 +0100
----------------------------------------------------------------------
.../java/org/apache/qpid/jms/JmsSession.java | 4 +-
.../org/apache/qpid/jms/message/JmsMessage.java | 2 +-
.../message/JmsMessagePropertyIntercepter.java | 4 +-
.../jms/message/facade/JmsMessageFacade.java | 11 ++-
.../jms/provider/amqp/AmqpFixedProducer.java | 2 +-
.../amqp/message/AmqpJmsMessageFacade.java | 19 +++++-
.../jms/integration/MessageIntegrationTest.java | 58 ++++++++++++++++
.../integration/ProducerIntegrationTest.java | 70 ++++++++++++++++++++
.../JmsMessagePropertyIntercepterTest.java | 6 +-
.../facade/test/JmsTestMessageFacade.java | 9 ++-
.../qpid/jms/producer/JmsProducerTest.java | 28 +++++++-
.../amqp/message/AmqpJmsMessageFacadeTest.java | 3 +
12 files changed, 199 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
index ca3bb26..8faafba 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
@@ -796,12 +796,12 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
// updating our own message instances, avoids using the interface
// in case the JMS 1.1 Message API is actually being used due to
// being on the classpath too.
- long deliveryTime = 0;
+ long deliveryTime = timeStamp;
if (hasDelay) {
deliveryTime = timeStamp + deliveryDelay;
}
- outbound.setJMSDeliveryTime(deliveryTime);
+ outbound.getFacade().setDeliveryTime(deliveryTime, hasDelay);
if(!isJmsMessage) {
// If the original was a foreign message, we still need to update it too.
setForeignMessageDeliveryTime(original, deliveryTime);
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessage.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessage.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessage.java
index c949c58..cd5f679 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessage.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessage.java
@@ -309,7 +309,7 @@ public class JmsMessage implements javax.jms.Message {
@Override
public void setJMSDeliveryTime(long deliveryTime) throws JMSException {
checkReadOnly();
- facade.setDeliveryTime(deliveryTime);
+ facade.setDeliveryTime(deliveryTime, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepter.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepter.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepter.java
index 65c8c9a..c9c6148 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepter.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepter.java
@@ -652,7 +652,7 @@ public class JmsMessagePropertyIntercepter {
if (rc == null) {
throw new JMSException("Property JMSDeliveryTime cannot be set from a " + value.getClass().getName() + ".");
}
- message.getFacade().setDeliveryTime(rc.longValue());
+ message.getFacade().setDeliveryTime(rc.longValue(), true);
}
@Override
@@ -662,7 +662,7 @@ public class JmsMessagePropertyIntercepter {
@Override
public void clearProperty(JmsMessage message) {
- message.getFacade().setDeliveryTime(0);
+ message.getFacade().setDeliveryTime(0, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java
index 048ff52..48f0524 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java
@@ -325,8 +325,10 @@ public interface JmsMessageFacade {
*
* @param deliveryTime
* the earliest time that the message should be made available for delivery.
+ * @param transmit
+ * whether to transmit an annotation containing the value (if non-zero)
*/
- void setDeliveryTime(long deliveryTime);
+ void setDeliveryTime(long deliveryTime, boolean transmit);
/**
* Gets the Destination value that was assigned to this message at the time it was
@@ -457,4 +459,11 @@ public interface JmsMessageFacade {
*/
Object encodeMessage();
+ /**
+ * Returns whether the delivery time is being transmitted, i.e. incorporates an actual delivery delay.
+ *
+ * @return true if delivery time is being transmitted as an annotation
+ */
+ boolean isDeliveryTimeTransmitted();
+
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java
index f654d48..e93d74e 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java
@@ -91,7 +91,7 @@ public class AmqpFixedProducer extends AmqpProducer {
request.onFailure(new IllegalStateException("The MessageProducer is closed"));
}
- if (!delayedDeliverySupported && envelope.getMessage().getJMSDeliveryTime() != 0) {
+ if (!delayedDeliverySupported && envelope.getMessage().getFacade().isDeliveryTimeTransmitted()) {
// Don't allow sends with delay if the remote has not said it can handle them
request.onFailure(new JMSException("Remote does not support delayed message delivery"));
} else if (getEndpoint().getCredit() <= 0) {
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacade.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacade.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacade.java
index 20dea80..7e514ba 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacade.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacade.java
@@ -70,6 +70,7 @@ public class AmqpJmsMessageFacade implements JmsMessageFacade {
private JmsDestination consumerDestination;
private Long syntheticExpiration;
+ private long syntheticDeliveryTime;
/**
* Used to record the value of JMS_AMQP_TTL property
@@ -106,6 +107,10 @@ public class AmqpJmsMessageFacade implements JmsMessageFacade {
if (absoluteExpiryTime == null && ttl != null) {
syntheticExpiration = System.currentTimeMillis() + ttl;
}
+
+ if (getMessageAnnotation(JMS_DELIVERY_TIME) == null) {
+ syntheticDeliveryTime = getTimestamp();
+ }
}
/**
@@ -250,6 +255,7 @@ public class AmqpJmsMessageFacade implements JmsMessageFacade {
target.connection = connection;
target.consumerDestination = consumerDestination;
target.syntheticExpiration = syntheticExpiration;
+ target.syntheticDeliveryTime = syntheticDeliveryTime;
target.userSpecifiedTTL = userSpecifiedTTL;
if (destination != null) {
@@ -548,18 +554,25 @@ public class AmqpJmsMessageFacade implements JmsMessageFacade {
return (long) deliveryTime;
}
- return 0l;
+ return syntheticDeliveryTime;
}
@Override
- public void setDeliveryTime(long deliveryTime) {
- if (deliveryTime != 0) {
+ public void setDeliveryTime(long deliveryTime, boolean transmit) {
+ if (deliveryTime != 0 && transmit) {
+ syntheticDeliveryTime = 0;
setMessageAnnotation(JMS_DELIVERY_TIME, deliveryTime);
} else {
+ syntheticDeliveryTime = deliveryTime;
removeMessageAnnotation(JMS_DELIVERY_TIME);
}
}
+ @Override
+ public boolean isDeliveryTimeTransmitted() {
+ return getMessageAnnotation(JMS_DELIVERY_TIME) != null;
+ }
+
/**
* Sets a value which will be used to override any ttl value that may otherwise be set
* based on the expiration value when sending the underlying AMQP message. A value of 0
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
index fa19b31..7560a24 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
@@ -2267,4 +2267,62 @@ public class MessageIntegrationTest extends QpidJmsTestCase
public void onException(Message message, Exception exception) {
}
}
+
+ //==== DeliveryTime Handling ====
+ //===============================
+
+ @Test(timeout = 20000)
+ public void testReceivedMessageWithDeliveryTimeAnnotation() throws Exception {
+ doReceivedMessageDeliveryTimeTestImpl(true);
+ }
+
+ @Test(timeout = 20000)
+ public void testReceivedMessageWithoutDeliveryTimeAnnotation() throws Exception {
+ doReceivedMessageDeliveryTimeTestImpl(false);
+ }
+
+ private void doReceivedMessageDeliveryTimeTestImpl(boolean setDeliveryTimeAnnotation)
+ throws JMSException, InterruptedException, Exception, IOException {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ Connection connection = testFixture.establishConnecton(testPeer);
+ connection.start();
+
+ testPeer.expectBegin();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ final long creationTime = System.currentTimeMillis();
+ final long deliveryTime = creationTime + 13526;
+
+ MessageAnnotationsDescribedType msgAnnotations = null;
+ if(setDeliveryTimeAnnotation) {
+ msgAnnotations = new MessageAnnotationsDescribedType();
+ msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_DELIVERY_TIME, deliveryTime);
+ }
+
+ PropertiesDescribedType props = new PropertiesDescribedType();
+ props.setTo("myAddress");
+ props.setMessageId("ID:myMessageIDString");
+ props.setCreationTime(new Date(creationTime));
+
+ testPeer.expectReceiverAttach();
+ testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, props, null, new AmqpValueDescribedType(null));
+ testPeer.expectDispositionThatIsAcceptedAndSettled();
+
+ MessageConsumer messageConsumer = session.createConsumer(queue);
+ Message receivedMessage = messageConsumer.receive(3000);
+ testPeer.waitForAllHandlersToComplete(3000);
+
+ testPeer.expectClose();
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(3000);
+
+ assertNotNull("should have recieved a message", receivedMessage);
+
+ long expectedDeliveryTime = setDeliveryTimeAnnotation ? deliveryTime : creationTime;
+ assertEquals("Unexpected delivery time", expectedDeliveryTime, receivedMessage.getJMSDeliveryTime());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
index 9ce040f..2175fff 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
@@ -19,6 +19,7 @@
package org.apache.qpid.jms.integration;
import static org.apache.qpid.jms.provider.amqp.AmqpSupport.DELAYED_DELIVERY;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.arrayContaining;
import static org.hamcrest.Matchers.both;
import static org.hamcrest.Matchers.equalTo;
@@ -28,6 +29,7 @@ import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -66,6 +68,7 @@ import org.apache.qpid.jms.JmsDefaultConnectionListener;
import org.apache.qpid.jms.JmsOperationTimedOutException;
import org.apache.qpid.jms.JmsSendTimedOutException;
import org.apache.qpid.jms.message.foreign.ForeignJmsMessage;
+import org.apache.qpid.jms.provider.amqp.AmqpSupport;
import org.apache.qpid.jms.provider.amqp.message.AmqpMessageIdHelper;
import org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport;
import org.apache.qpid.jms.test.QpidJmsTestCase;
@@ -93,6 +96,7 @@ import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.hamcrest.Matcher;
import org.hamcrest.MatcherAssert;
import org.hamcrest.Matchers;
+import org.hamcrest.core.CombinableMatcher;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -2677,6 +2681,72 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test(timeout = 20000)
+ public void testSendingMessageSetsJMSDeliveryTimeWithDelay() throws Exception {
+ doSendingMessageSetsJMSDeliveryTimeTestImpl(true);
+ }
+
+ @Test(timeout = 20000)
+ public void testSendingMessageSetsJMSDeliveryTimeWithoutDelay() throws Exception {
+ doSendingMessageSetsJMSDeliveryTimeTestImpl(false);
+ }
+
+ private void doSendingMessageSetsJMSDeliveryTimeTestImpl(boolean deliveryDelay) throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ // add connection capability to indicate server support for DELAYED-DELIVERY
+ Connection connection = testFixture.establishConnecton(testPeer, new Symbol[]{ DELAYED_DELIVERY });
+
+ testPeer.expectBegin();
+ testPeer.expectSenderAttach();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ String queueName = "myQueue";
+ Queue queue = session.createQueue(queueName);
+ MessageProducer producer = session.createProducer(queue);
+
+ int delay = 0;
+ if(deliveryDelay) {
+ delay = 123456;
+ producer.setDeliveryDelay(delay);
+ }
+
+ // Create matcher to expect the DeliveryTime to be set to a value
+ // representing 'now' [+ delivery-delay], within a upper delta for execution time.
+ long deliveryTimeLower = System.currentTimeMillis();
+ long deliveryTimeUpper = deliveryTimeLower + delay + 3000;
+ Matcher<Long> inRange = both(greaterThanOrEqualTo(deliveryTimeLower)).and(lessThanOrEqualTo(deliveryTimeUpper));
+ Symbol DELIVERY_TIME = Symbol.valueOf("x-opt-delivery-time");
+
+ String text = "myMessage";
+ TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
+ messageMatcher.setHeadersMatcher(new MessageHeaderSectionMatcher(true));
+ MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
+ if(deliveryDelay) {
+ msgAnnotationsMatcher.withEntry(DELIVERY_TIME, inRange);
+ }
+ messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
+ messageMatcher.setPropertiesMatcher(new MessagePropertiesSectionMatcher(true));
+ messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(text));
+ testPeer.expectTransfer(messageMatcher);
+
+ Message message = session.createTextMessage(text);
+
+ producer.send(message);
+
+ testPeer.expectClose();
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+
+ if (!deliveryDelay) {
+ assertFalse("Message should not have delivery time annotation",
+ msgAnnotationsMatcher.keyExistsInReceivedAnnotations(DELIVERY_TIME));
+ }
+
+ assertThat(message.getJMSDeliveryTime(), inRange);
+ }
+ }
+
private class TestJmsCompletionListener implements CompletionListener {
private final CountDownLatch completed;
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepterTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepterTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepterTest.java
index 055602d..4a0c6f2 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepterTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/JmsMessagePropertyIntercepterTest.java
@@ -1815,7 +1815,7 @@ public class JmsMessagePropertyIntercepterTest {
JmsMessage message = Mockito.mock(JmsMapMessage.class);
Mockito.when(message.getFacade()).thenReturn(facade);
JmsMessagePropertyIntercepter.setProperty(message, JMS_DELIVERYTIME, 65536L);
- Mockito.verify(facade).setDeliveryTime(65536L);
+ Mockito.verify(facade).setDeliveryTime(65536L, true);
}
@Test
@@ -1884,8 +1884,8 @@ public class JmsMessagePropertyIntercepterTest {
JmsMessage message = Mockito.mock(JmsMapMessage.class);
Mockito.when(message.getFacade()).thenReturn(facade);
JmsMessagePropertyIntercepter.clearProperties(message, true);
- Mockito.verify(facade, Mockito.never()).setDeliveryTime(0);
+ Mockito.verify(facade, Mockito.never()).setDeliveryTime(0, true);
JmsMessagePropertyIntercepter.clearProperties(message, false);
- Mockito.verify(facade).setDeliveryTime(0);
+ Mockito.verify(facade).setDeliveryTime(0, true);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java
index d4b1cfd..a8f0eab 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java
@@ -61,6 +61,7 @@ public class JmsTestMessageFacade implements JmsMessageFacade {
protected Object messageId;
protected long expiration;
protected long deliveryTime;
+ protected boolean deliveryTimeTransmitted;
protected long timestamp;
protected String correlationId;
protected boolean persistent = true;
@@ -290,8 +291,14 @@ public class JmsTestMessageFacade implements JmsMessageFacade {
}
@Override
- public void setDeliveryTime(long deliveryTime) {
+ public void setDeliveryTime(long deliveryTime, boolean transmit) {
this.deliveryTime = deliveryTime;
+ this.deliveryTimeTransmitted = transmit;
+ }
+
+ @Override
+ public boolean isDeliveryTimeTransmitted() {
+ return deliveryTimeTransmitted;
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/test/java/org/apache/qpid/jms/producer/JmsProducerTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/producer/JmsProducerTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/producer/JmsProducerTest.java
index 52c66ee..b00a26e 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/producer/JmsProducerTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/producer/JmsProducerTest.java
@@ -18,6 +18,10 @@
*/
package org.apache.qpid.jms.producer;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.both;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -57,6 +61,7 @@ import org.apache.qpid.jms.message.JmsMessage;
import org.apache.qpid.jms.message.JmsOutboundMessageDispatch;
import org.apache.qpid.jms.meta.JmsSessionId;
import org.apache.qpid.jms.provider.mock.MockRemotePeer;
+import org.hamcrest.Matcher;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -938,18 +943,35 @@ public class JmsProducerTest extends JmsConnectionTestSupport {
private void doTestSendAppliesDeliveryDelayMessageBody(Class<?> bodyType) throws JMSException {
JMSProducer producer = context.createProducer();
+ // Create matcher to expect the DeliveryTime to be set to a value
+ // representing 'now', within a upper delta for execution time.
+ long deliveryTimeLower = System.currentTimeMillis();
+ long deliveryTimeUpper = deliveryTimeLower + 3000;
+ Matcher<Long> inRange = both(greaterThanOrEqualTo(deliveryTimeLower)).and(lessThanOrEqualTo(deliveryTimeUpper));
+
sendWithBodyOfType(producer, bodyType);
+
JmsOutboundMessageDispatch envelope = remotePeer.getLastReceivedMessage();
assertNotNull(envelope);
JmsMessage message = envelope.getMessage();
- assertTrue(message.getJMSDeliveryTime() == 0);
+ assertThat(message.getJMSDeliveryTime(), inRange);
+
+ // Repeat with a non-zero delay
+ int deliveryDelay = 123456;
+ producer.setDeliveryDelay(deliveryDelay);
+
+ // Create matcher to expect the DeliveryTime to be set to a value
+ // representing 'now' + delivery-delay, within a upper delta for execution time.
+ deliveryTimeLower = System.currentTimeMillis();
+ deliveryTimeUpper = deliveryTimeLower + deliveryDelay + 3000;
+ inRange = both(greaterThanOrEqualTo(deliveryTimeLower)).and(lessThanOrEqualTo(deliveryTimeUpper));
- producer.setDeliveryDelay(2000);
sendWithBodyOfType(producer, bodyType);
+
envelope = remotePeer.getLastReceivedMessage();
assertNotNull(envelope);
message = envelope.getMessage();
- assertFalse(message.getJMSDeliveryTime() == 0);
+ assertThat(message.getJMSDeliveryTime(), inRange);
}
@Test
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/70fe1b88/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java
index 782ab0c..de2171a 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java
@@ -2192,6 +2192,7 @@ public class AmqpJmsMessageFacadeTest extends AmqpJmsMessageTypesTestCase {
source.setRedeliveryCount(12);
source.setTimestamp(150L);
source.setUserId("Cookie-Monster");
+ source.setDeliveryTime(123456, false);
source.setProperty("APP-Prop-1", "APP-Prop-1-Value");
source.setProperty("APP-Prop-2", "APP-Prop-2-Value");
@@ -2215,6 +2216,8 @@ public class AmqpJmsMessageFacadeTest extends AmqpJmsMessageTypesTestCase {
assertEquals(source.getRedeliveryCount(), copy.getRedeliveryCount());
assertEquals(source.getTimestamp(), copy.getTimestamp());
assertEquals(source.getUserId(), copy.getUserId());
+ assertEquals(source.getDeliveryTime(), copy.getDeliveryTime());
+
// There should be two since none of the extended options were set
assertEquals(2, copy.getPropertyNames().size());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org