QPID-2881

When a topic based destination is used in creating a consumer, the code looks to see if the address is already resolved.

If it is then it could mean two things.
  1. A consumer is being recreated after failover.
  2. This destination was used previously to create a consumer or a producer. In this case we assume it was used for a consumer.

For case #1, we need to preserve the queue name. The reason being if it was either a durable subscription or if it was a named queue (name property specified in link) then we need to maintain the same name.

For case #2 we need to ensure that each consumer gets it's own queue,hence we clone the destination object and the queue name is set to null
so name is auto-generated.

I have removed the incorrect logic (in BasicMessageConsumer_0_10) of looking at the durable property as one factor in deciding if it's case #1.
A durable subscription does use a named queue, therefore it is not nessacery to look at the durable property in order to identify it as case #1.
I have made modifications to createDurableTopic method in AMQTopic to set the queue name in the link properties.

The above changes enable a user to create more than one consumer (each with it's own unique queue) when using the same destination created off an addressing string that has durable set to true.


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

2 files changed, 8 insertions, 6 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java b/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
index 3f80d01811..6217cb534a 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
@@ -103,9 +103,14 @@ public class AMQTopic extends AMQDestination implements Topic
             try
             {
                 AMQTopic t = new AMQTopic(topic.getAddress());
-                t.setQueueName(getDurableTopicQueueName(subscriptionName, connection));
+                AMQShortString queueName = getDurableTopicQueueName(subscriptionName, connection);
+                // link is never null if dest was created using an address string.
+                t.getLink().setName(queueName.asString());               
                 t.getSourceNode().setAutoDelete(false);
                 t.getSourceNode().setDurable(true);
+                
+                // The legacy fields are also populated just in case.
+                t.setQueueName(queueName);
                 t.setAutoDelete(false);
                 t.setDurable(true);
                 return t;
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
index a0d29093eb..35c0c66c7f 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
@@ -117,18 +117,15 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<UnprocessedM
         }
 
         if (destination.isAddressResolved() && AMQDestination.TOPIC_TYPE == destination.getAddressType()) 
-        {
-            
-            boolean durable = destination.getSourceNode() != null && destination.getSourceNode().isDurable();
+        {            
             boolean namedQueue = destination.getLink() != null && destination.getLink().getName() != null ; 
             
-            if (!durable && !namedQueue)
+            if (!namedQueue)
             {
                 _destination = destination.copyDestination();
                 _destination.setQueueName(null);
             }
         }
-        
     }