QPID-8141: [JMS AMQP 0-x] Avoid repeated exchange.declare when publishing when BURL...
authorKeith Wall <kwall@apache.org>
Wed, 2 May 2018 21:32:30 +0000 (22:32 +0100)
committerKeith Wall <kwall@apache.org>
Wed, 2 May 2018 21:42:43 +0000 (22:42 +0100)
client/src/main/java/org/apache/qpid/client/AMQSession.java
systests/src/test/java/org/apache/qpid/systest/destination/AddressBasedDestinationTest.java

index cba764e..b5d4870 100644 (file)
@@ -30,7 +30,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -150,8 +150,8 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
      */
     protected final boolean DAEMON_DISPATCHER_THREAD = Boolean.getBoolean(ClientProperties.DAEMON_DISPATCHER);
 
-    private final Set<AMQDestination>
-            _resolvedDestinations = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<AMQDestination, Boolean>()));
+    private final Map<AMQDestination, AMQDestination>
+            _resolvedDestinations = Collections.synchronizedMap(new WeakHashMap<AMQDestination, AMQDestination>());
 
     private final long _dispatcherShutdownTimeoutMs;
 
@@ -661,7 +661,7 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
 
     void setResolved(final AMQDestination dest)
     {
-        _resolvedDestinations.add(dest);
+        _resolvedDestinations.put(dest, dest);
     }
 
     void setUnresolved(final AMQDestination dest)
@@ -676,30 +676,22 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
 
     boolean isResolved(final AMQDestination dest)
     {
-        if (!_resolvedDestinations.contains(dest))
+        AMQDestination resolvedDest = _resolvedDestinations.get(dest);
+        if (resolvedDest == dest)
         {
-            return false;
-        }
-
-        if (dest.getAddressType() == AMQDestination.QUEUE_TYPE)
-        {
-            // verify legacy fields are set
-            return dest.getQueueName() != null
-                   && dest.getQueueName().equals(dest.getAddressName())
-                   && dest.getExchangeName() != null
-                   && dest.getExchangeClass() != null
-                   && dest.getRoutingKey() != null;
+            return true;
         }
-        else if (dest.getAddressType() == AMQDestination.TOPIC_TYPE)
+        else if (resolvedDest == null)
         {
-            // verify legacy fields are set
-            return dest.getExchangeName() != null
-                   && dest.getExchangeName().equals(dest.getAddressName())
-                   && dest.getExchangeClass() != null
-                   && (dest.getSubject() == null
-                        || (dest.getSubject() != null && dest.getSubject().equals(dest.getRoutingKey())));
+            return false;
         }
-        return false;
+
+        // verify legacy fields are equal
+        return Objects.equals(dest.getQueueName(), resolvedDest.getQueueName()) &&
+               Objects.equals(dest.getExchangeName(), resolvedDest.getExchangeName()) &&
+               Objects.equals(dest.getExchangeClass(), resolvedDest.getExchangeClass()) &&
+               Objects.equals(dest.getRoutingKey(), resolvedDest.getRoutingKey()) &&
+               Objects.equals(dest.getSubject(), resolvedDest.getSubject());
     }
 
     public abstract int resolveAddressType(AMQDestination dest) throws QpidException;
index 395b410..1665283 100644 (file)
@@ -26,6 +26,7 @@ import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -373,16 +374,21 @@ public class AddressBasedDestinationTest extends JmsTestBase
         }
     }
 
+    /** QPID-8141 - Publishing to a duplicately declared queue silently dropped messages.  */
     @Test
-    public void ensureQueueDestinationAlwaysResolved() throws Exception
+    public void publishToDuplicatelyDeclaredQueue() throws Exception
     {
         String address = String.format("ADDR:%s; {create: always, node: {type: queue}}", getTestName());
 
         Session session = _connection.createSession(true, Session.SESSION_TRANSACTED);
 
         MessageProducer producer = session.createProducer(null);
-        producer.send(session.createQueue(address), session.createTextMessage("A"));
-        producer.send(session.createQueue(address), session.createTextMessage("B"));
+        Queue queue = session.createQueue(address);
+        Queue dupQueue = session.createQueue(address);
+        assertNotSame(queue, dupQueue);
+
+        producer.send(queue, session.createTextMessage("A"));
+        producer.send(dupQueue, session.createTextMessage("B"));
         session.commit();
 
         MessageConsumer consumer = session.createConsumer(session.createQueue(address));
@@ -397,17 +403,20 @@ public class AddressBasedDestinationTest extends JmsTestBase
         assertEquals("Unexpected content of message B", "B", ((TextMessage) messageB).getText());
     }
 
-
     @Test
-    public void ensureTopicDestinationAlwaysResolved() throws Exception
+    public void publishToDuplicatelyDeclaredTopic() throws Exception
     {
         String address = String.format("ADDR:amq.topic/%s; {node: {type: topic}}", getTestName());
         Session session = _connection.createSession(true, Session.SESSION_TRANSACTED);
         MessageConsumer consumer = session.createConsumer(session.createTopic(address));
 
         MessageProducer producer = session.createProducer(null);
-        producer.send(session.createTopic(address), session.createTextMessage("A"));
-        producer.send(session.createTopic(address), session.createTextMessage("B"));
+        Topic topic = session.createTopic(address);
+        Topic dupTopic = session.createTopic(address);
+        assertNotSame(topic, dupTopic);
+
+        producer.send(topic, session.createTextMessage("A"));
+        producer.send(dupTopic, session.createTextMessage("B"));
         session.commit();
 
         Message messageA = consumer.receive(getReceiveTimeout());