You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2017/01/09 13:11:35 UTC
svn commit: r1777959 - in /qpid/java/trunk:
broker-core/src/main/java/org/apache/qpid/server/model/
systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/
Author: kwall
Date: Mon Jan 9 13:11:35 2017
New Revision: 1777959
URL: http://svn.apache.org/viewvc?rev=1777959&view=rev
Log:
QPID-7568: [Java Broker] [JMS 2.0 - Delivery Delay] Add system tests for feature
Added:
qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/
qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/DeliveryDelayTest.java
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java?rev=1777959&r1=1777958&r2=1777959&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java Mon Jan 9 13:11:35 2017
@@ -42,6 +42,7 @@ public interface Exchange<X extends Exch
String ALTERNATE_EXCHANGE = "alternateExchange";
String DURABLE_BINDINGS = "durableBindings";
+ String UNROUTABLE_MESSAGE_BEHAVIOUR = "unroutableMessageBehaviour";
enum UnroutableMessageBehaviour
{
Added: qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/DeliveryDelayTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/DeliveryDelayTest.java?rev=1777959&view=auto
==============================================================================
--- qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/DeliveryDelayTest.java (added)
+++ qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/deliverydelay/DeliveryDelayTest.java Mon Jan 9 13:11:35 2017
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.systests.jms_2_0.deliverydelay;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.InvalidDestinationRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSProducer;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class DeliveryDelayTest extends QpidBrokerTestCase
+{
+ private static final int DELIVERY_DELAY = 3000;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ setTestSystemProperty("virtualhost.housekeepingCheckPeriod", "100");
+ super.setUp();
+ }
+
+ public void testDeliveryDelay() throws Exception
+ {
+ ConnectionFactory connectionFactory = getConnectionFactory();
+ try (JMSContext context = connectionFactory.createContext(GUEST_USERNAME, GUEST_PASSWORD);
+ Connection utilityConnection = connectionFactory.createConnection(GUEST_USERNAME, GUEST_PASSWORD))
+ {
+ Destination queue = createQueue(utilityConnection, getTestQueueName(), true);
+
+ final AtomicLong messageReceiptTime = new AtomicLong();
+ final CountDownLatch receivedLatch = new CountDownLatch(1);
+ context.createConsumer(queue).setMessageListener(new MessageListener()
+ {
+ @Override
+ public void onMessage(final Message message)
+ {
+ messageReceiptTime.set(System.currentTimeMillis());
+ receivedLatch.countDown();
+ }
+ });
+
+ JMSProducer producer = context.createProducer().setDeliveryDelay(DELIVERY_DELAY);
+
+ final long messageSentTime = System.currentTimeMillis();
+ producer.send(queue, "delayed message");
+
+ final boolean messageArrived = receivedLatch.await(DELIVERY_DELAY * 3, TimeUnit.MILLISECONDS);
+ assertTrue("Delayed message did not arrive within expected period", messageArrived);
+ final long actualDelay = messageReceiptTime.get() - messageSentTime;
+ assertTrue(String.format("Message was not delayed by sufficient time (%d). Actual delay (%d)",
+ DELIVERY_DELAY, actualDelay),
+ actualDelay >= DELIVERY_DELAY);
+ }
+ }
+
+ /**
+ * The target queue, which is addressed directly by the client, does not have
+ * holdsOnPublish turned on. The Broker must reject the message.
+ */
+ public void testDeliveryDelayNotSupportedByQueue_MessageRejected() throws Exception
+ {
+ ConnectionFactory connectionFactory = getConnectionFactory();
+ try (JMSContext context = connectionFactory.createContext(GUEST_USERNAME, GUEST_PASSWORD);
+ Connection utilityConnection = connectionFactory.createConnection(GUEST_USERNAME, GUEST_PASSWORD))
+ {
+ Destination queue = createQueue(utilityConnection, getTestQueueName(), false);
+ JMSProducer producer = context.createProducer().setDeliveryDelay(DELIVERY_DELAY);
+
+ try
+ {
+ producer.send(queue, "message");
+ fail("Exception not thrown");
+ }
+ catch (InvalidDestinationRuntimeException e)
+ {
+ // PASS
+ }
+ }
+ }
+
+ /**
+ * The client sends a messagge to a fanout exchange instance which is bound to a queue with
+ * holdsOnPublish turned off. The Broker must reject the message.
+ */
+ public void testDeliveryDelayNotSupportedByQueueViaExchange_MessageRejected() throws Exception
+ {
+ ConnectionFactory connectionFactory = getConnectionFactory();
+ try (JMSContext context = connectionFactory.createContext(GUEST_USERNAME, GUEST_PASSWORD);
+ Connection utilityConnection = connectionFactory.createConnection(GUEST_USERNAME, GUEST_PASSWORD))
+ {
+ String testQueueName = getTestQueueName();
+ String testExchangeName = getTestName() + "_exch";
+
+ Destination consumeDest = createQueue(utilityConnection, testQueueName, false);
+ Destination publishDest = createExchange(utilityConnection, testExchangeName);
+ bindQueueToExchange(utilityConnection, testExchangeName, testQueueName);
+
+
+ JMSConsumer consumer = context.createConsumer(consumeDest);
+ JMSProducer producer = context.createProducer();
+
+ producer.send(publishDest, "message without delivery delay");
+
+ Message message = consumer.receive(getReceiveTimeout());
+ assertNotNull("Message published without delivery delay not received", message);
+
+ producer.setDeliveryDelay(DELIVERY_DELAY);
+
+ try
+ {
+ producer.send(publishDest, "message with delivery delay");
+ fail("Exception not thrown");
+ }
+ catch (InvalidDestinationRuntimeException e)
+ {
+ // PASS
+ }
+ }
+ }
+
+ private Destination createQueue(Connection utilityConnection, String queueName, boolean holdsOnPublish) throws Exception
+ {
+ try (Session session = utilityConnection.createSession(Session.SESSION_TRANSACTED))
+ {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(org.apache.qpid.server.model.Queue.HOLD_ON_PUBLISH_ENABLED, holdsOnPublish);
+ createEntityUsingAmqpManagement(queueName, session, "org.apache.qpid.Queue", attributes);
+ return session.createQueue(queueName);
+ }
+ }
+
+ private Destination createExchange(Connection utilityConnection, String exchangeName) throws Exception
+ {
+ try (Session session = utilityConnection.createSession(Session.SESSION_TRANSACTED))
+ {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(org.apache.qpid.server.model.Exchange.UNROUTABLE_MESSAGE_BEHAVIOUR, "REJECT");
+ createEntityUsingAmqpManagement(exchangeName, session, "org.apache.qpid.FanoutExchange", attributes);
+ return session.createQueue(exchangeName);
+ }
+ }
+
+ private void bindQueueToExchange(Connection utilityConnection,
+ String exchangeName,
+ String queueName) throws Exception
+ {
+ try (Session session = utilityConnection.createSession(Session.SESSION_TRANSACTED))
+ {
+ final Map<String, Object> arguments = new HashMap<>();
+ arguments.put("destination", queueName);
+ arguments.put("bindingKey", queueName);
+ performOperationUsingAmqpManagement(exchangeName, "bind", session, "org.apache.qpid.FanoutExchange", arguments);
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org