Do not declare reply-to destination for consumed messages

This commit instructs to not declare a reply-to RMQDestination for
consumed messages. This avoids trying to create a second time a
temporary reply-to destination (reply-to destinations are supposed to be
created by a RPC client before sending the request).

This happens to also fix #47.

Fixes #69

(cherry picked from commit 09282e4)

Conflicts:
	src/test/java/com/rabbitmq/integration/tests/RpcIT.java
	src/test/java/com/rabbitmq/integration/tests/RpcWithAmqpDirectReplyIT.java

Author: acogoluegnes

src/main/java/com/rabbitmq/jms/client/RMQMessage.java

@@ -908,6 +908,7 @@ static RMQMessage convertJmsMessage(RMQSession session, RMQDestination dest, Get
         message.setReadonly(true);                                              // Set readOnly - mandatory for received messages
 
         maybeSetupDirectReplyTo(message, response.getProps().getReplyTo());
+        doNotDeclareReplyToDestination(message);
 
         return message;
     }
@@ -927,6 +928,7 @@ private static RMQMessage convertAmqpMessage(RMQSession session, RMQDestination
             message.setReadonly(true);                                              // Set readOnly - mandatory for received messages
 
             maybeSetupDirectReplyTo(message, response.getProps().getReplyTo());
+            doNotDeclareReplyToDestination(message);
 
             return message;
         } catch (IOException x) {
@@ -955,6 +957,23 @@ private static void maybeSetupDirectReplyTo(RMQMessage message, String replyTo)
         }
     }
 
+    /**
+     * Indicates to not declare a reply-to {@link RMQDestination}.
+     * <p>
+     * This is used for inbound messages, when the reply-to destination
+     * is supposed to be already created. This avoids trying to create
+     * a second time a temporary queue.
+     *
+     * @param message
+     * @throws JMSException
+     * @since 1.11.0
+     */
+    private static void doNotDeclareReplyToDestination(RMQMessage message) throws JMSException {
+        if (message.getJMSReplyTo() != null && message.getJMSReplyTo() instanceof RMQDestination) {
+            ((RMQDestination) message.getJMSReplyTo()).setDeclared(true);
+        }
+    }
+
     /**
      * Generate the headers for an AMQP message.
      * <p>

src/test/java/com/rabbitmq/integration/tests/RpcIT.java

@@ -31,7 +31,6 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
 
 /**
  *
@@ -94,16 +93,7 @@ public void tearDown() throws Exception {
     }
 
     @Test
-    public void noResponseWhenServerTriesToRecreateTemporaryResponseQueue() throws Exception {
-        setupRpcServer();
-
-        String messageContent = UUID.randomUUID().toString();
-        Message response = doRpc(messageContent);
-        assertNull(response);
-    }
-
-    @Test
-    public void responseOkWhenServerDoesNotRecreateTemporaryResponseQueue() throws Exception {
+    public void rpc() throws Exception {
         setupRpcServer(destinationAlreadyDeclaredForRpcResponse());
 
         String messageContent = UUID.randomUUID().toString();
@@ -148,15 +138,6 @@ void setupRpcServer(SendingContextConsumer sendingContextConsumer) throws Except
         rpcServer = new RpcServer(serverConnection);
     }
 
-    void setupRpcServer() throws Exception {
-        setupRpcServer(new SendingContextConsumer() {
-
-            @Override
-            public void accept(SendingContext ctx) {
-            }
-        });
-    }
-
     private static class RpcServer {
 
         Session session;

src/test/java/com/rabbitmq/integration/tests/RpcWithAmqpDirectReplyIT.java

@@ -1,10 +1,7 @@
 /* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.integration.tests;
 
-import com.rabbitmq.jms.admin.RMQConnectionFactory;
 import com.rabbitmq.jms.admin.RMQDestination;
-import com.rabbitmq.jms.client.SendingContext;
-import com.rabbitmq.jms.client.SendingContextConsumer;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -52,19 +49,6 @@ protected static void drainQueue(Session session, Queue queue) throws Exception
         }
     }
 
-    static SendingContextConsumer destinationAlreadyDeclaredForRpcResponse() {
-        return new SendingContextConsumer() {
-
-            @Override
-            public void accept(SendingContext ctx) throws JMSException {
-                if (ctx.getMessage().getJMSCorrelationID() != null && ctx.getDestination() instanceof RMQDestination) {
-                    RMQDestination destination = (RMQDestination) ctx.getDestination();
-                    destination.setDeclared(true);
-                }
-            }
-        };
-    }
-
     @Before
     public void init() throws Exception {
         ConnectionFactory connectionFactory = AbstractTestConnectionFactory.getTestConnectionFactory()
@@ -93,7 +77,7 @@ public void tearDown() throws Exception {
 
     @Test
     public void responseOkWhenServerDoesNotRecreateTemporaryResponseQueue() throws Exception {
-        setupRpcServer(destinationAlreadyDeclaredForRpcResponse());
+        setupRpcServer();
 
         String messageContent = UUID.randomUUID().toString();
         Message response = doRpc(messageContent);
@@ -128,10 +112,9 @@ public void onMessage(Message msg) {
         return queue.poll(2, TimeUnit.SECONDS);
     }
 
-    void setupRpcServer(SendingContextConsumer sendingContextConsumer) throws Exception {
-        RMQConnectionFactory connectionFactory = (RMQConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory()
+    void setupRpcServer() throws Exception {
+        ConnectionFactory connectionFactory = AbstractTestConnectionFactory.getTestConnectionFactory()
             .getConnectionFactory();
-        connectionFactory.setSendingContextConsumer(sendingContextConsumer);
         serverConnection = connectionFactory.createConnection();
         serverConnection.start();
         Session session = serverConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

src/test/java/com/rabbitmq/integration/tests/SendingContextConsumerIT.java

@@ -0,0 +1,79 @@
+/* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
+package com.rabbitmq.integration.tests;
+
+import com.rabbitmq.jms.admin.RMQConnectionFactory;
+import com.rabbitmq.jms.client.SendingContext;
+import com.rabbitmq.jms.client.SendingContextConsumer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ *
+ */
+public class SendingContextConsumerIT {
+
+    private static final String QUEUE_NAME = "test.queue." + SendingContextConsumerIT.class.getCanonicalName();
+    final AtomicInteger sentCount = new AtomicInteger(0);
+    Connection connection;
+
+    protected static void drainQueue(Session session, Queue queue) throws Exception {
+        MessageConsumer receiver = session.createConsumer(queue);
+        Message msg = receiver.receiveNoWait();
+        while (msg != null) {
+            msg = receiver.receiveNoWait();
+        }
+    }
+
+    @Before
+    public void init() throws Exception {
+        RMQConnectionFactory connectionFactory = (RMQConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory()
+            .getConnectionFactory();
+        connectionFactory.setSendingContextConsumer(new SendingContextConsumer() {
+
+            @Override
+            public void accept(SendingContext ctx) {
+                sentCount.incrementAndGet();
+            }
+        });
+        connection = connectionFactory.createConnection();
+        connection.start();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        if (connection != null) {
+            connection.close();
+        }
+        com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();
+        com.rabbitmq.client.Connection c = cf.newConnection();
+        try {
+            c.createChannel().queueDelete(QUEUE_NAME);
+        } finally {
+            c.close();
+        }
+    }
+
+    @Test
+    public void sendingContextConsumerShouldBeCalledWhenSendingMessage() throws Exception {
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        TextMessage message = session.createTextMessage("hello");
+        MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
+        int initialCount = sentCount.get();
+        producer.send(message);
+        assertEquals(initialCount + 1, sentCount.get());
+        producer.send(message);
+        assertEquals(initialCount + 2, sentCount.get());
+    }
+}