QPID-509 Mandatory messages not returned outside a transaction. They are now.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@550849 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 183 insertions, 119 deletions

diff --git a/java/systests/src/main/java/org/apache/qpid/server/exchange/ImmediateMessageTest.java b/java/systests/src/main/java/org/apache/qpid/server/exchange/ImmediateMessageTest.java
index c8a244743e..048fcfb0b3 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/exchange/ImmediateMessageTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/exchange/ImmediateMessageTest.java
@@ -25,6 +25,7 @@ import junit.framework.TestCase;
 import org.apache.log4j.NDC;

 

 import org.apache.qpid.client.AMQNoConsumersException;

+import org.apache.qpid.client.AMQNoRouteException;

 import org.apache.qpid.client.AMQSession;

 import org.apache.qpid.client.transport.TransportConnection;

 import static org.apache.qpid.server.exchange.MessagingTestConfigProperties.*;

@@ -113,70 +114,70 @@ public class ImmediateMessageTest extends TestCase
         testClients.testNoExceptions(testProps);

     }

 

-    /** Check that an immediate message results in no consumers code, not using transactions, when no consumer is connected. */

-    public void test_QPID_517_ImmediateFailsNoConsumerNoTxP2P() throws Exception

+    /** Check that an immediate message results in no consumers code, not using transactions, when a consumer is disconnected. */

+    public void test_QPID_517_ImmediateFailsConsumerDisconnectedNoTxP2P() throws Exception

     {

         // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, false);

         testProps.setProperty(PUBSUB_PROPNAME, false);

 

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

-

         PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

+        // Disconnect the consumer.

+        testClients.getReceiver().getConsumer().close();

+

         // Send one message and get a linked no consumers exception.

         testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

     }

 

-    /** Check that an immediate message results in no consumers code, upon transaction commit, when a consumer is connected. */

-    public void test_QPID_517_ImmediateFailsNoConsumerTxP2P() throws Exception

+    /** Check that an immediate message results in no consumers code, in a transaction, when a consumer is disconnected. */

+    public void test_QPID_517_ImmediateFailsConsumerDisconnectedTxP2P() throws Exception

     {

         // Ensure transactional sessions are on.

         testProps.setProperty(TRANSACTED_PROPNAME, true);

         testProps.setProperty(PUBSUB_PROPNAME, false);

 

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

-

         PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

+        // Disconnect the consumer.

+        testClients.getReceiver().getConsumer().close();

+

         // Send one message and get a linked no consumers exception.

         testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

     }

 

-    /** Check that an immediate message results in no consumers code, not using transactions, when a consumer is disconnected. */

-    public void test_QPID_517_ImmediateFailsConsumerDisconnectedNoTxP2P() throws Exception

+    /** Check that an immediate message results in no consumers code, not using transactions, when no consumer is connected. */

+    public void test_QPID_517_ImmediateFailsNoRouteNoTxP2P() throws Exception

     {

         // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, false);

         testProps.setProperty(PUBSUB_PROPNAME, false);

 

-        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

 

-        // Disconnect the consumer.

-        testClients.getReceiver().getConsumer().close();

+        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

         // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

     }

 

-    /** Check that an immediate message results in no consumers code, in a transaction, when a consumer is disconnected. */

-    public void test_QPID_517_ImmediateFailsConsumerDisconnectedTxP2P() throws Exception

+    /** Check that an immediate message results in no consumers code, upon transaction commit, when a consumer is connected. */

+    public void test_QPID_517_ImmediateFailsNoRouteTxP2P() throws Exception

     {

         // Ensure transactional sessions are on.

         testProps.setProperty(TRANSACTED_PROPNAME, true);

         testProps.setProperty(PUBSUB_PROPNAME, false);

 

-        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

 

-        // Disconnect the consumer.

-        testClients.getReceiver().getConsumer().close();

+        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

         // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

     }

 

     /** Check that an immediate message is sent succesfully not using transactions when a consumer is connected. */

@@ -205,70 +206,76 @@ public class ImmediateMessageTest extends TestCase
         testClients.testNoExceptions(testProps);

     }

 

-    /** Check that an immediate message results in no consumers code, not using transactions, when no consumer is connected. */

-    public void test_QPID_517_ImmediateFailsNoConsumerNoTxPubSub() throws Exception

+    /** Check that an immediate message results in no consumers code, not using transactions, when a consumer is disconnected. */

+    public void test_QPID_517_ImmediateFailsConsumerDisconnectedNoTxPubSub() throws Exception

     {

         // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, false);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

+        // Use durable subscriptions, so that the route remains open with no subscribers.

+        testProps.setProperty(DURABLE_SUBSCRIPTION_PROPNAME, true);

 

         PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

+        // Disconnect the consumer.

+        testClients.getReceiver().getConsumer().close();

+

         // Send one message and get a linked no consumers exception.

         testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

     }

 

-    /** Check that an immediate message results in no consumers code, upon transaction commit, when a consumer is connected. */

-    public void test_QPID_517_ImmediateFailsNoConsumerTxPubSub() throws Exception

+    /** Check that an immediate message results in no consumers code, in a transaction, when a consumer is disconnected. */

+    public void test_QPID_517_ImmediateFailsConsumerDisconnectedTxPubSub() throws Exception

     {

         // Ensure transactional sessions are on.

         testProps.setProperty(TRANSACTED_PROPNAME, true);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

+        // Use durable subscriptions, so that the route remains open with no subscribers.

+        testProps.setProperty(DURABLE_SUBSCRIPTION_PROPNAME, true);

 

         PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

+        // Disconnect the consumer.

+        testClients.getReceiver().getConsumer().close();

+

         // Send one message and get a linked no consumers exception.

         testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

     }

 

-    /** Check that an immediate message results in no consumers code, not using transactions, when a consumer is disconnected. */

-    public void test_QPID_517_ImmediateFailsConsumerDisconnectedNoTxPubSub() throws Exception

+    /** Check that an immediate message results in no consumers code, not using transactions, when no consumer is connected. */

+    public void test_QPID_517_ImmediateFailsNoRouteNoTxPubSub() throws Exception

     {

         // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, false);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

-        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

 

-        // Disconnect the consumer.

-        testClients.getReceiver().getConsumer().close();

+        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

         // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

     }

 

-    /** Check that an immediate message results in no consumers code, in a transaction, when a consumer is disconnected. */

-    public void test_QPID_517_ImmediateFailsConsumerDisconnectedTxPubSub() throws Exception

+    /** Check that an immediate message results in no consumers code, upon transaction commit, when a consumer is connected. */

+    public void test_QPID_517_ImmediateFailsNoRouteTxPubSub() throws Exception

     {

         // Ensure transactional sessions are on.

         testProps.setProperty(TRANSACTED_PROPNAME, true);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

-        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

 

-        // Disconnect the consumer.

-        testClients.getReceiver().getConsumer().close();

+        PublisherReceiver testClients = PublisherReceiverImpl.connectClients(testProps);

 

         // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoConsumersException.class);

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

     }

 

     protected void setUp() throws Exception

@@ -425,6 +432,14 @@ public class ImmediateMessageTest extends TestCase
         }

     }

 

+    public static class MessageMonitor implements MessageListener

+    {

+        public void onMessage(Message message)

+        {

+            log.debug("public void onMessage(Message message): called");

+        }

+    }

+

     /**

      * Establishes a JMS connection using a properties file and qpids built in JNDI implementation. This is a simple

      * convenience method for code that does anticipate handling connection failures. All exceptions that indicate

@@ -505,13 +520,14 @@ public class ImmediateMessageTest extends TestCase
             int ackMode = messagingProps.getPropertyAsInteger(ACK_MODE_PROPNAME);

             boolean useTopics = messagingProps.getPropertyAsBoolean(PUBSUB_PROPNAME);

             boolean transactional = messagingProps.getPropertyAsBoolean(TRANSACTED_PROPNAME);

+            boolean durableSubscription = messagingProps.getPropertyAsBoolean(DURABLE_SUBSCRIPTION_PROPNAME);

 

             // Check if any Qpid/AMQP specific flags or options need to be set.

             boolean immediate = messagingProps.getPropertyAsBoolean(IMMEDIATE_PROPNAME);

             boolean mandatory = messagingProps.getPropertyAsBoolean(MANDATORY_PROPNAME);

             boolean needsQpidOptions = immediate | mandatory;

 

-            log.debug("ackMode = " + ackMode);

+            /*log.debug("ackMode = " + ackMode);

             log.debug("useTopics = " + useTopics);

             log.debug("destinationSendRoot = " + destinationSendRoot);

             log.debug("destinationReceiveRoot = " + destinationReceiveRoot);

@@ -522,7 +538,7 @@ public class ImmediateMessageTest extends TestCase
             log.debug("transactional = " + transactional);

             log.debug("immediate = " + immediate);

             log.debug("mandatory = " + mandatory);

-            log.debug("needsQpidOptions = " + needsQpidOptions);

+            log.debug("needsQpidOptions = " + needsQpidOptions);*/

 

             // Create connection, sessions and producer/consumer pairs on each session.

             Connection connection = createConnection(messagingProps);

@@ -535,7 +551,7 @@ public class ImmediateMessageTest extends TestCase
             Session receiverSession = connection.createSession(transactional, ackMode);

 

             Destination publisherProducerDestination =

-                useTopics ? publisherSession.createTopic(destinationSendRoot)

+                useTopics ? (Destination) publisherSession.createTopic(destinationSendRoot)

                           : publisherSession.createQueue(destinationSendRoot);

 

             MessageProducer publisherProducer =

@@ -548,13 +564,29 @@ public class ImmediateMessageTest extends TestCase
                 createPublisherConsumer

                 ? publisherSession.createConsumer(publisherSession.createQueue(destinationReceiveRoot)) : null;

 

+            if (publisherConsumer != null)

+            {

+                publisherConsumer.setMessageListener(new MessageMonitor());

+            }

+

             MessageProducer receiverProducer =

                 createReceiverProducer ? receiverSession.createProducer(receiverSession.createQueue(destinationReceiveRoot))

                                        : null;

 

+            Destination receiverConsumerDestination =

+                useTopics ? (Destination) receiverSession.createTopic(destinationSendRoot)

+                          : receiverSession.createQueue(destinationSendRoot);

+

             MessageConsumer receiverConsumer =

-                createReceiverConsumer ? receiverSession.createConsumer(receiverSession.createQueue(destinationSendRoot))

-                                       : null;

+                createReceiverConsumer

+                ? ((durableSubscription && useTopics)

+                    ? receiverSession.createDurableSubscriber((Topic) receiverConsumerDestination, "testsub")

+                    : receiverSession.createConsumer(receiverConsumerDestination)) : null;

+

+            if (receiverConsumer != null)

+            {

+                receiverConsumer.setMessageListener(new MessageMonitor());

+            }

 

             // Start listening for incoming messages.

             connection.start();

@@ -578,7 +610,8 @@ public class ImmediateMessageTest extends TestCase
 

     public static Message createTestMessage(ProducerConsumerPair client, ParsedProperties testProps) throws JMSException

     {

-        return client.getSession().createMessage();

+        return client.getSession().createTextMessage("Hello");

+            // return client.getSession().createMessage();

     }

 

     /**

@@ -868,9 +901,7 @@ public class ImmediateMessageTest extends TestCase
         public static PublisherReceiver connectClients(ParsedProperties testProps)

         {

             // Create a standard publisher/receiver test client pair on a shared connection, individual sessions.

-            PublisherReceiver testClients = createPublisherReceiverPairSharedConnection(testProps);

-

-            return testClients;

+            return createPublisherReceiverPairSharedConnection(testProps);

         }

     }

 

diff --git a/java/systests/src/main/java/org/apache/qpid/server/exchange/MandatoryMessageTest.java b/java/systests/src/main/java/org/apache/qpid/server/exchange/MandatoryMessageTest.java
index 8fbda6f54b..09a32aa3eb 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/exchange/MandatoryMessageTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/exchange/MandatoryMessageTest.java
@@ -91,42 +91,6 @@ public class MandatoryMessageTest extends TestCase
         testClients.testNoExceptions(testProps);

     }

 

-    /** Check that an mandatory message results in no route code, not using transactions, when no consumer is connected. */

-    public void test_QPID_508_MandatoryFailsNoRouteNoTxP2P() throws Exception

-    {

-        // Ensure transactional sessions are off.

-        testProps.setProperty(TRANSACTED_PROPNAME, false);

-        testProps.setProperty(PUBSUB_PROPNAME, false);

-

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

-

-        ImmediateMessageTest.PublisherReceiver testClients =

-            ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

-

-        // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

-    }

-

-    /** Check that an mandatory message results in no route code, upon transaction commit, when a consumer is connected. */

-    public void test_QPID_508_MandatoryFailsNoRouteTxP2P() throws Exception

-    {

-        // Ensure transactional sessions are on.

-        testProps.setProperty(TRANSACTED_PROPNAME, true);

-        testProps.setProperty(PUBSUB_PROPNAME, false);

-

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

-

-        ImmediateMessageTest.PublisherReceiver testClients =

-            ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

-

-        // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

-    }

-

     /**

      * Check that a mandatory message is sent succesfully, not using transactions, when a consumer is disconnected but

      * the route exists.

@@ -167,68 +131,68 @@ public class MandatoryMessageTest extends TestCase
         testClients.testNoExceptions(testProps);

     }

 

-    /** Check that an mandatory message is sent succesfully not using transactions when a consumer is connected. */

-    public void test_QPID_508_MandatoryOkNoTxPubSub() throws Exception

+    /** Check that an mandatory message results in no route code, not using transactions, when no consumer is connected. */

+    public void test_QPID_508_MandatoryFailsNoRouteNoTxP2P() throws Exception

     {

         // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, false);

-        testProps.setProperty(PUBSUB_PROPNAME, true);

+        testProps.setProperty(PUBSUB_PROPNAME, false);

+

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

 

         ImmediateMessageTest.PublisherReceiver testClients =

             ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

 

-        // Send one message with no errors.

-        testClients.testNoExceptions(testProps);

+        // Send one message and get a linked no consumers exception.

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

     }

 

-    /** Check that an mandatory message is committed succesfully in a transaction when a consumer is connected. */

-    public void test_QPID_508_MandatoryOkTxPubSub() throws Exception

+    /** Check that an mandatory message results in no route code, upon transaction commit, when a consumer is connected. */

+    public void test_QPID_508_MandatoryFailsNoRouteTxP2P() throws Exception

     {

-        // Ensure transactional sessions are off.

+        // Ensure transactional sessions are on.

         testProps.setProperty(TRANSACTED_PROPNAME, true);

-        testProps.setProperty(PUBSUB_PROPNAME, true);

+        testProps.setProperty(PUBSUB_PROPNAME, false);

+

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

 

         ImmediateMessageTest.PublisherReceiver testClients =

             ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

 

-        // Send one message with no errors.

-        testClients.testNoExceptions(testProps);

+        // Send one message and get a linked no consumers exception.

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

     }

 

-    /** Check that an mandatory message results in no route code, not using transactions, when no consumer is connected. */

-    public void test_QPID_508_MandatoryFailsNoRouteNoTxPubSub() throws Exception

+    /** Check that an mandatory message is sent succesfully not using transactions when a consumer is connected. */

+    public void test_QPID_508_MandatoryOkNoTxPubSub() throws Exception

     {

         // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, false);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

-

         ImmediateMessageTest.PublisherReceiver testClients =

             ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

 

-        // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

+        // Send one message with no errors.

+        testClients.testNoExceptions(testProps);

     }

 

-    /** Check that an mandatory message results in no route code, upon transaction commit, when a consumer is connected. */

-    public void test_QPID_508_MandatoryFailsNoRouteTxPubSub() throws Exception

+    /** Check that an mandatory message is committed succesfully in a transaction when a consumer is connected. */

+    public void test_QPID_508_MandatoryOkTxPubSub() throws Exception

     {

-        // Ensure transactional sessions are on.

+        // Ensure transactional sessions are off.

         testProps.setProperty(TRANSACTED_PROPNAME, true);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

-        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

-        // collect its messages).

-        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

-

         ImmediateMessageTest.PublisherReceiver testClients =

             ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

 

-        // Send one message and get a linked no consumers exception.

-        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

+        // Send one message with no errors.

+        testClients.testNoExceptions(testProps);

     }

 

     /**

@@ -241,6 +205,9 @@ public class MandatoryMessageTest extends TestCase
         testProps.setProperty(TRANSACTED_PROPNAME, false);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

+        // Use durable subscriptions, so that the route remains open with no subscribers.

+        testProps.setProperty(DURABLE_SUBSCRIPTION_PROPNAME, true);

+

         ImmediateMessageTest.PublisherReceiver testClients =

             ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

 

@@ -261,6 +228,9 @@ public class MandatoryMessageTest extends TestCase
         testProps.setProperty(TRANSACTED_PROPNAME, true);

         testProps.setProperty(PUBSUB_PROPNAME, true);

 

+        // Use durable subscriptions, so that the route remains open with no subscribers.

+        testProps.setProperty(DURABLE_SUBSCRIPTION_PROPNAME, true);

+

         ImmediateMessageTest.PublisherReceiver testClients =

             ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

 

@@ -271,6 +241,42 @@ public class MandatoryMessageTest extends TestCase
         testClients.testNoExceptions(testProps);

     }

 

+    /** Check that an mandatory message results in no route code, not using transactions, when no consumer is connected. */

+    public void test_QPID_508_MandatoryFailsNoRouteNoTxPubSub() throws Exception

+    {

+        // Ensure transactional sessions are off.

+        testProps.setProperty(TRANSACTED_PROPNAME, false);

+        testProps.setProperty(PUBSUB_PROPNAME, true);

+

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

+

+        ImmediateMessageTest.PublisherReceiver testClients =

+            ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

+

+        // Send one message and get a linked no consumers exception.

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

+    }

+

+    /** Check that an mandatory message results in no route code, upon transaction commit, when a consumer is connected. */

+    public void test_QPID_508_MandatoryFailsNoRouteTxPubSub() throws Exception

+    {

+        // Ensure transactional sessions are on.

+        testProps.setProperty(TRANSACTED_PROPNAME, true);

+        testProps.setProperty(PUBSUB_PROPNAME, true);

+

+        // Set up the messaging topology so that only the publishers producer is bound (do not set up the receiver to

+        // collect its messages).

+        testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, false);

+

+        ImmediateMessageTest.PublisherReceiver testClients =

+            ImmediateMessageTest.PublisherReceiverImpl.connectClients(testProps);

+

+        // Send one message and get a linked no consumers exception.

+        testClients.testWithAssertions(testProps, AMQNoRouteException.class);

+    }

+

     protected void setUp() throws Exception

     {

         NDC.push(getName());

diff --git a/java/systests/src/main/java/org/apache/qpid/server/exchange/MessagingTestConfigProperties.java b/java/systests/src/main/java/org/apache/qpid/server/exchange/MessagingTestConfigProperties.java
index 9c8cefc492..b584c8c80b 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/exchange/MessagingTestConfigProperties.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/exchange/MessagingTestConfigProperties.java
@@ -1,3 +1,23 @@
+/*

+ *

+ * 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.server.exchange;

 

 import org.apache.qpid.jms.Session;

@@ -167,6 +187,12 @@ public class MessagingTestConfigProperties
     /** Defines the default message acknowledgement mode. */

     public static final int ACK_MODE_DEFAULT = Session.AUTO_ACKNOWLEDGE;

 

+    /** Holds the name of the property to get the durable subscriptions flag from, when doing pub/sub messaging. */

+    public static final String DURABLE_SUBSCRIPTION_PROPNAME = "durableSubscription";

+

+    /** Defines the default value of the durable subscriptions flag. */

+    public static final boolean DURABLE_SUBSCRIPTION_DEFAULT = false;

+

     // ======================  Qpid Options and Flags ================================

 

     /** Holds the name of the property to set the exclusive flag from. */

@@ -272,6 +298,7 @@ public class MessagingTestConfigProperties
         defaults.setPropertyIfNull(TX_BATCH_SIZE_PROPNAME, TX_BATCH_SIZE_DEFAULT);

         defaults.setPropertyIfNull(DURABLE_DESTS_PROPNAME, DURABLE_DESTS_DEFAULT);

         defaults.setPropertyIfNull(ACK_MODE_PROPNAME, ACK_MODE_DEFAULT);

+        defaults.setPropertyIfNull(DURABLE_SUBSCRIPTION_PROPNAME, DURABLE_SUBSCRIPTION_DEFAULT);

         defaults.setPropertyIfNull(MAX_PENDING_PROPNAME, MAX_PENDING_DEFAULT);

         defaults.setPropertyIfNull(PREFECTH_PROPNAME, PREFETCH_DEFAULT);

         defaults.setPropertyIfNull(NO_LOCAL_PROPNAME, NO_LOCAL_DEFAULT);