QPID-3539: NoLocal should occur at the connection level and not the session level (0-10). Also fixed typing issue that prevented Java Broker understanding the no-local queue argument when encoded as String (as sent by Python tests).

Applied patch from Oleksandr Rudyy<orudyy@gmail.com> and myself.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1200335 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 143 insertions, 0 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
index 9ea116ae53..d3d9cf2984 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
@@ -841,4 +841,147 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
             e.printStackTrace();
         }
     }
+
+    /**
+     * Tests that a subscriber created on a same <i>session</i> as producer with
+     * no local true does not receive messages.
+     */
+    public void testNoLocalOnSameSession() throws Exception
+    {
+        Connection connection = getConnection();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Topic topic = session.createTopic(getTestQueueName());
+        MessageProducer producer = session.createProducer(topic);
+        TopicSubscriber subscriber =  null;
+        try
+        {
+            subscriber = session.createDurableSubscriber(topic, getTestName(), null, true);
+            connection.start();
+
+            producer.send(createNextMessage(session, 1));
+
+            Message m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT);
+            assertNull("Unexpected message received", m);
+        }
+        finally
+        {
+            session.unsubscribe(getTestName());
+        }
+    }
+
+
+    /**
+     * Tests that a subscriber created on a same <i>connection</i> but separate
+     * <i>sessionM</i> as producer with no local true does not receive messages.
+     */
+    public void testNoLocalOnSameConnection() throws Exception
+    {
+        Connection connection = getConnection();
+
+        Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Topic topic = consumerSession.createTopic(getTestQueueName());
+        MessageProducer producer = producerSession.createProducer(topic);
+
+        TopicSubscriber subscriber =  null;
+        try
+        {
+            subscriber = consumerSession.createDurableSubscriber(topic, getTestName(), null, true);
+            connection.start();
+
+            producer.send(createNextMessage(producerSession, 1));
+
+            Message m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT);
+            assertNull("Unexpected message received", m);
+        }
+        finally
+        {
+            consumerSession.unsubscribe(getTestName());
+        }
+    }
+
+    /**
+     * Tests that if no-local is in use, that the messages are delivered when
+     * the client reconnects.
+     *
+     * Currently fails on the Java Broker due to QPID-3605.
+     */
+    public void testNoLocalMessagesNotDeliveredAfterReconnection() throws Exception
+    {
+        Connection connection = getConnection();
+
+        Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Topic topic = consumerSession.createTopic(getTestQueueName());
+        MessageProducer producer = producerSession.createProducer(topic);
+
+        TopicSubscriber subscriber =  null;
+        try
+        {
+            subscriber = consumerSession.createDurableSubscriber(topic, getTestName(), null, true);
+            connection.start();
+
+            producer.send(createNextMessage(producerSession, 1));
+
+            Message m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT);
+            assertNull("Unexpected message received", m);
+
+            connection.close();
+
+            connection = getConnection();
+
+            consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            subscriber = consumerSession.createDurableSubscriber(topic, getTestName(), null, true);
+            connection.start();
+            m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT);
+            assertNull("Message should not be received on a new connection", m);
+        }
+        finally
+        {
+            consumerSession.unsubscribe(getTestName());
+        }
+    }
+
+    /**
+     * Tests that messages are delivered normally to a subscriber on a separate connection despite
+     * the use of durable subscriber with no-local on the first connection.
+     */
+    public void testNoLocalSubscriberAndSubscriberOnSeparateConnection() throws Exception
+    {
+        Connection noLocalConnection = getConnection();
+        Connection connection = getConnection();
+
+        String noLocalSubId1 = getTestName() + "subId1";
+        String subId = getTestName() + "subId2";
+
+        Session noLocalSession = noLocalConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Topic noLocalTopic = noLocalSession.createTopic(getTestQueueName());
+
+        Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Topic topic = consumerSession.createTopic(getTestQueueName());
+
+        TopicSubscriber noLocalSubscriber =  null;
+        TopicSubscriber subscriber =  null;
+        try
+        {
+            MessageProducer producer = noLocalSession.createProducer(noLocalTopic);
+            noLocalSubscriber = noLocalSession.createDurableSubscriber(noLocalTopic, noLocalSubId1, null, true);
+            subscriber = consumerSession.createDurableSubscriber(topic, subId, null, true);
+            noLocalConnection.start();
+            connection.start();
+
+            producer.send(createNextMessage(noLocalSession, 1));
+
+            Message m1 = noLocalSubscriber.receive(NEGATIVE_RECEIVE_TIMEOUT);
+            assertNull("Subscriber on nolocal connection should not receive message", m1);
+
+            Message m2 = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT);
+            assertNotNull("Subscriber on non-nolocal connection should receive message", m2);
+        }
+        finally
+        {
+            noLocalSession.unsubscribe(noLocalSubId1);
+            consumerSession.unsubscribe(subId);
+        }
+    }
 }