diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java index f80fffda..01f12c90 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java +++ b/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.Map; 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 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 1.3 */ -public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, RoutingConnectionFactory { +public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, RoutingConnectionFactory, + InitializingBean { private final Map targetConnectionFactories = new ConcurrentHashMap(); @@ -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. *

The key can be of arbitrary type; this class implements the @@ -58,6 +64,7 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact 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 abstract class AbstractRoutingConnectionFactory implements ConnectionFact */ public void setDefaultTargetConnectionFactory(ConnectionFactory defaultTargetConnectionFactory) { this.defaultTargetConnectionFactory = defaultTargetConnectionFactory; + checkConfirmsAndReturns(defaultTargetConnectionFactory); } /** @@ -93,9 +101,37 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact 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(); } /** diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java index 7b48f2dd..db455e0d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java +++ b/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.junit.jupiter.api.Test; 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 @@ import com.rabbitmq.client.Envelope; * @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 class RabbitTemplatePublisherCallbacksIntegrationTests2 { 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.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; diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index baf4705a..69decc0f 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -691,6 +691,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 <>. + [[queue-affinity]] ===== Queue Affinity and the `LocalizedQueueConnectionFactory` @@ -1196,7 +1199,7 @@ 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, 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 return message is set before the future is set with the `ack`. +It is guaranteed that the returned message is set before the future is set with the `ack`. See also <> for a simpler mechanism for waiting for publisher confirms.