GH-1289: Confirms and Returns with Routing CF

Resolves #1289

`RoutingConnectionFactory` did not support correlated confirms or returns.

Target factories (and default) must have the same settings.

**cherry-pick to 2.2.x, 2.1.x**

# Conflicts:
#	src/reference/asciidoc/amqp.adoc

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.springframework.amqp.AmqpException;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -35,7 +36,8 @@
  * @author Gary Russell
  * @since 1.3
  */
-public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, RoutingConnectionFactory {
+public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, RoutingConnectionFactory,
+		InitializingBean {
 
 	private final Map<Object, ConnectionFactory> targetConnectionFactories =
 			new ConcurrentHashMap<Object, ConnectionFactory>();
@@ -46,6 +48,10 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
 
 	private boolean lenientFallback = true;
 
+	private Boolean confirms;
+
+	private Boolean returns;
+
 	/**
 	 * Specify the map of target ConnectionFactories, with the lookup key as key.
 	 * <p>The key can be of arbitrary type; this class implements the
@@ -58,6 +64,7 @@ public void setTargetConnectionFactories(Map<Object, ConnectionFactory> targetCo
 		Assert.noNullElements(targetConnectionFactories.values().toArray(),
 				"'targetConnectionFactories' cannot have null values.");
 		this.targetConnectionFactories.putAll(targetConnectionFactories);
+		targetConnectionFactories.values().stream().forEach(cf -> checkConfirmsAndReturns(cf));
 	}
 
 	/**
@@ -69,6 +76,7 @@ public void setTargetConnectionFactories(Map<Object, ConnectionFactory> targetCo
 	 */
 	public void setDefaultTargetConnectionFactory(ConnectionFactory defaultTargetConnectionFactory) {
 		this.defaultTargetConnectionFactory = defaultTargetConnectionFactory;
+		checkConfirmsAndReturns(defaultTargetConnectionFactory);
 	}
 
 	/**
@@ -93,9 +101,37 @@ public boolean isLenientFallback() {
 		return this.lenientFallback;
 	}
 
+	@Override
+	public boolean isPublisherConfirms() {
+		return this.confirms;
+	}
+
+	@Override
+	public boolean isPublisherReturns() {
+		return this.returns;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.confirms, "At least one target factory (or default) is required");
+	}
+
+	private void checkConfirmsAndReturns(ConnectionFactory cf) {
+		if (this.confirms == null) {
+			this.confirms = cf.isPublisherConfirms();
+		}
+		if (this.returns == null) {
+			this.returns = cf.isPublisherReturns();
+		}
+		Assert.isTrue(this.confirms.booleanValue() == cf.isPublisherConfirms(),
+				"Target connection factories must have the same setting for publisher confirms");
+		Assert.isTrue(this.returns.booleanValue() == cf.isPublisherReturns(),
+				"Target connection factories must have the same setting for publisher returns");
+	}
+
 	@Override
 	public Connection createConnection() throws AmqpException {
-		return this.determineTargetConnectionFactory().createConnection();
+		return determineTargetConnectionFactory().createConnection();
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,6 +27,8 @@
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory;
 import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
 import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 
@@ -39,36 +41,38 @@
  * @since 1.6
  *
  */
-@RabbitAvailable(queues = RabbitTemplatePublisherCallbacksIntegrationTests2.ROUTE)
+@RabbitAvailable(queues = { RabbitTemplatePublisherCallbacksIntegrationTests2.ROUTE,
+		RabbitTemplatePublisherCallbacksIntegrationTests2.ROUTE2 })
 public class RabbitTemplatePublisherCallbacksIntegrationTests2 {
 
 	public static final String ROUTE = "test.queue.RabbitTemplatePublisherCallbacksIntegrationTests2";
 
+	public static final String ROUTE2 = "test.queue.RabbitTemplatePublisherCallbacksIntegrationTests2.route";
+
 	private CachingConnectionFactory connectionFactoryWithConfirmsEnabled;
 
 	private RabbitTemplate templateWithConfirmsEnabled;
 
 	@BeforeEach
-	public void create() {
+	void create() {
 		connectionFactoryWithConfirmsEnabled = new CachingConnectionFactory();
 		connectionFactoryWithConfirmsEnabled.setHost("localhost");
-		// When using publisher confirms, the cache size needs to be large enough
-		// otherwise channels can be closed before confirms are received.
 		connectionFactoryWithConfirmsEnabled.setChannelCacheSize(100);
 		connectionFactoryWithConfirmsEnabled.setPort(BrokerTestUtils.getPort());
 		connectionFactoryWithConfirmsEnabled.setPublisherConfirmType(ConfirmType.CORRELATED);
+		connectionFactoryWithConfirmsEnabled.setPublisherReturns(true);
 		templateWithConfirmsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsEnabled);
 	}
 
 	@AfterEach
-	public void cleanUp() {
+	void cleanUp() {
 		if (connectionFactoryWithConfirmsEnabled != null) {
 			connectionFactoryWithConfirmsEnabled.destroy();
 		}
 	}
 
 	@Test
-	public void test36Methods() throws Exception {
+	void test36Methods() throws Exception {
 		this.templateWithConfirmsEnabled.convertAndSend(ROUTE, "foo");
 		this.templateWithConfirmsEnabled.convertAndSend(ROUTE, "foo");
 		assertMessageCountEquals(2L);
@@ -91,6 +95,51 @@ public void handleDelivery(String ag, Envelope envelope, BasicProperties propert
 		assertMessageCountEquals(0L);
 	}
 
+	@Test
+	void routingWithConfirmsNoListener() throws Exception {
+		routingWithConfirms(false);
+	}
+
+	@Test
+	void routingWithConfirmsListener() throws Exception {
+		routingWithConfirms(true);
+	}
+
+	private void routingWithConfirms(boolean listener) throws Exception {
+		CountDownLatch latch = new CountDownLatch(1);
+		SimpleRoutingConnectionFactory rcf = new SimpleRoutingConnectionFactory();
+		rcf.setDefaultTargetConnectionFactory(this.connectionFactoryWithConfirmsEnabled);
+		this.templateWithConfirmsEnabled.setConnectionFactory(rcf);
+		if (listener) {
+			this.templateWithConfirmsEnabled.setConfirmCallback((correlationData, ack, cause) -> {
+				latch.countDown();
+			});
+		}
+		this.templateWithConfirmsEnabled.setMandatory(true);
+		CorrelationData corr = new CorrelationData();
+		this.templateWithConfirmsEnabled.convertAndSend("", ROUTE2, "foo", corr);
+		assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
+		if (listener) {
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+		corr = new CorrelationData();
+		this.templateWithConfirmsEnabled.convertAndSend("", "bad route", "foo", corr);
+		assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
+		assertThat(corr.getReturnedMessage()).isNotNull();
+	}
+
+	@Test
+	void routingWithSimpleConfirms() throws Exception {
+		SimpleRoutingConnectionFactory rcf = new SimpleRoutingConnectionFactory();
+		rcf.setDefaultTargetConnectionFactory(this.connectionFactoryWithConfirmsEnabled);
+		this.templateWithConfirmsEnabled.setConnectionFactory(rcf);
+	 	assertThat(this.templateWithConfirmsEnabled.<Boolean>invoke(template -> {
+				template.convertAndSend("", ROUTE2, "foo");
+				template.waitForConfirmsOrDie(10_000);
+				return true;
+	 	})).isTrue();
+	}
+
 	private void assertMessageCountEquals(long wanted) throws InterruptedException {
 		long messageCount = determineMessageCount();
 		int n = 0;

src/reference/asciidoc/amqp.adoc

@@ -630,6 +630,9 @@ Doing so enables, for example, listening to queues with the same name but in a d
 
 For example, with lookup key qualifier `thing1` and a container listening to queue `thing2`, the lookup key you could register the target connection factory with could be `thing1[thing2]`.
 
+IMPORTANT: The target (and default, if provided) connection factories must have the same settings for publisher confirms and returns.
+See <<cf-pub-conf-ret>>.
+
 [[queue-affinity]]
 ===== Queue Affinity and the `LocalizedQueueConnectionFactory`
 
@@ -1127,8 +1130,8 @@ The `Confirm` object is a simple bean with 2 properties: `ack` and `reason` (for
 The reason is not populated for broker-generated `nack` instances.
 It is populated for `nack` instances generated by the framework (for example, closing the connection while `ack` instances are outstanding).
 
-In addition, when both confirms and returns are enabled, the `CorrelationData` is populated with the returned message.
-It is guaranteed that this occurs before the future is set with the `ack`.
+In addition, when both confirms and returns are enabled, the `CorrelationData` is populated with the returned message, as long as the `CorrelationData` has a unique `id`; this is always the case, by default, starting with version 2.3.
+It is guaranteed that the returned message is set before the future is set with the `ack`.
 
 See also <<scoped-operations>> for a simpler mechanism for waiting for publisher confirms.