QPID-929 : Exchange.Declare being sent prior to every message when publishing to explicit destination

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

3 files changed, 28 insertions, 11 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java b/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
index cc5af07b20..c795e2bf43 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
@@ -53,7 +53,9 @@ public abstract class AMQDestination implements Destination, Referenceable
     private String _url;
     private AMQShortString _urlAsShortString;
 
-    private boolean _validated;
+    private boolean _checkedForQueueBinding;
+
+    private boolean _exchangeExistsChecked;
 
     private byte[] _byteEncoding;
     private static final int IS_DURABLE_MASK = 0x1;
@@ -202,14 +204,25 @@ public abstract class AMQDestination implements Destination, Referenceable
 
     }
 
-    public boolean isValidated()
+    public boolean isCheckedForQueueBinding()
+    {
+        return _checkedForQueueBinding;
+    }
+
+    public void setCheckedForQueueBinding(boolean checkedForQueueBinding)
+    {
+        _checkedForQueueBinding = checkedForQueueBinding;
+    }
+
+
+    public boolean isExchangeExistsChecked()
     {
-        return _validated;
+        return _exchangeExistsChecked;
     }
 
-    public void setValidated(boolean validated)
+    public void setExchangeExistsChecked(final boolean exchangeExistsChecked)
     {
-        _validated = validated;
+        _exchangeExistsChecked = exchangeExistsChecked;
     }
 
     public String toURL()
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
index ae71846870..191d967396 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
@@ -140,6 +140,7 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
 
     private void declareDestination(AMQDestination destination)
     {
+
         ExchangeDeclareBody body = getSession().getMethodRegistry().createExchangeDeclareBody(_session.getTicket(),
                                                                                               destination.getExchangeName(),
                                                                                               destination.getExchangeClass(),
@@ -410,7 +411,12 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
                                    + ((destination != null) ? destination.getClass() : null));
         }
 
-        declareDestination((AMQDestination) destination);
+        AMQDestination amqDestination = (AMQDestination) destination;
+        if(!amqDestination.isExchangeExistsChecked())
+        {
+            declareDestination(amqDestination);
+            amqDestination.setExchangeExistsChecked(true);
+        }
     }
 
     protected void sendImpl(AMQDestination destination, Message message, int deliveryMode, int priority, long timeToLive,
diff --git a/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java b/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
index 493e2b5ec0..27783bcacf 100644
--- a/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
+++ b/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
@@ -22,11 +22,9 @@
 package org.apache.qpid.client;
 
 import javax.jms.Destination;
-import javax.jms.IllegalStateException;
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
 import javax.jms.Message;
-import javax.jms.MessageProducer;
 import javax.jms.Queue;
 import javax.jms.QueueSender;
 
@@ -200,19 +198,19 @@ public class QueueSenderAdapter implements QueueSender
         }
 
         AMQDestination destination = (AMQDestination) queue;
-        if (!destination.isValidated() && checkQueueBeforePublish())
+        if (!destination.isCheckedForQueueBinding() && checkQueueBeforePublish())
         {
 
             if (_delegate.getSession().isStrictAMQP())
             {
                 _delegate._logger.warn("AMQP does not support destination validation before publish, ");
-                destination.setValidated(true);
+                destination.setCheckedForQueueBinding(true);
             }
             else
             {
                 if (_delegate.isBound(destination))
                 {
-                    destination.setValidated(true);
+                    destination.setCheckedForQueueBinding(true);
                 }
                 else
                 {