GH-846: Fix send and receive with confirms

Fixes #846

Send and receive with a direct container (the default) fails on the second
operation because the template is not registered as a listener with the
callback channel.

`doSendAndReceiveWithDirect` does not invoke `doSendAndReceiveAsListener`
within `execute` because the channel is obtained from the container.

The first send and receive succeeds because `execute` is invoked once in the
`useDirectReplyTo` which determines whether the broker supports direct
reply-to.

`addListener()` is called from `execute()` after a channel has been received
from the cache.

Fix is to call `addListener()` from `doSendAndReceiveWithDirect`.

**cherry-pick to 2.0.x**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -1769,8 +1769,11 @@ private Message doSendAndReceiveWithDirect(String exchange, String routingKey, M
 		}
 		ChannelHolder channelHolder = container.getChannelHolder();
 		try {
-			return doSendAndReceiveAsListener(exchange, routingKey, message, correlationData,
-					channelHolder.getChannel());
+			Channel channel = channelHolder.getChannel();
+			if (this.confirmsOrReturnsCapable) {
+				addListener(channel);
+			}
+			return doSendAndReceiveAsListener(exchange, routingKey, message, correlationData, channel);
 		}
 		catch (Exception e) {
 			container.releaseConsumerFor(channelHolder, false, null);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java

@@ -40,13 +40,16 @@
  *
  */
 @RabbitAvailable(queues = { RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE1,
-		RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE2 })
+		RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE2,
+		RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE3 })
 public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
 
 	public static final String QUEUE1 = "synthetic.nack";
 
 	public static final String QUEUE2 = "defer.close";
 
+	public static final String QUEUE3 = "confirm.send.receive";
+
 	@Test
 	public void testRepublishOnNackThreadNoExchange() throws Exception {
 		CachingConnectionFactory cf = new CachingConnectionFactory(
@@ -123,6 +126,25 @@ public void testDeferredChannelCacheAck() throws Exception {
 		cf.destroy();
 	}
 
+	@Test
+	public void testTwoSendsAndReceivesDRTMLC() throws Exception {
+		CachingConnectionFactory cf = new CachingConnectionFactory(
+				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		cf.setPublisherConfirms(true);
+		RabbitTemplate template = new RabbitTemplate(cf);
+		template.setReplyTimeout(0);
+		final CountDownLatch confirmLatch = new CountDownLatch(2);
+		template.setConfirmCallback((cd, a, c) -> {
+			confirmLatch.countDown();
+		});
+		template.convertSendAndReceive("", QUEUE3, "foo", new MyCD("foo"));
+		template.convertSendAndReceive("", QUEUE3, "foo", new MyCD("foo")); // listener not registered
+		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(template.receive(QUEUE3, 10_000)).isNotNull();
+		assertThat(template.receive(QUEUE3, 10_000)).isNotNull();
+	}
+
+
 	private static class MyCD extends CorrelationData {
 
 		private final String payload;