From acb56833d1af567c8d1a839ca92c97f5a8fe7d33 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 19 Aug 2019 11:21:21 -0400 Subject: [PATCH] GH-1067: Add enum for publisher confirm type Resolves https://github.com/spring-projects/spring-amqp/issues/1067 * * Refactor deprecated setters to use new setter * * Fix this.publisherConnectionFactory typo --- .../config/ConnectionFactoryParser.java | 1 + .../connection/CachingConnectionFactory.java | 86 ++++++++++++++----- .../amqp/rabbit/config/spring-rabbit-2.2.xsd | 21 ++++- .../amqp/rabbit/AsyncRabbitTemplateTests.java | 3 +- .../config/ConnectionFactoryParserTests.java | 7 +- .../CachingConnectionFactoryTests.java | 11 ++- .../core/RabbitTemplateIntegrationTests.java | 3 +- ...atePublisherCallbacksIntegrationTests.java | 21 ++--- ...tePublisherCallbacksIntegrationTests2.java | 3 +- ...tePublisherCallbacksIntegrationTests3.java | 9 +- .../core/SimplePublisherConfirmsTests.java | 5 +- .../ConnectionFactoryParserTests-context.xml | 1 + src/reference/asciidoc/amqp.adoc | 12 +-- src/reference/asciidoc/whats-new.adoc | 2 + 14 files changed, 130 insertions(+), 55 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java index ecc24c87..25473e2c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java @@ -111,6 +111,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, FACTORY_TIMEOUT, "channelCheckoutTimeout"); NamespaceUtils.setValueIfAttributeDefined(builder, element, CONNECTION_LIMIT); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, "connection-name-strategy"); + NamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-type", "publisherConfirmType"); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 31b1bf5c..5972f3b1 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -129,14 +129,41 @@ public class CachingConnectionFactory extends AbstractConnectionFactory * The cache mode. */ public enum CacheMode { + /** * Cache channels - single connection. */ CHANNEL, + /** * Cache connections and channels within each connection. */ CONNECTION + + } + + /** + * The type of publisher confirms to use. + */ + public enum ConfirmType { + + /** + * Use {@code RabbitTemplate#waitForConfirms()} (or {@code waitForConfirmsOrDie()} + * within scoped operations. + */ + SIMPLE, + + /** + * Use with {@code CorrelationData} to correlate confirmations with sent + * messsages. + */ + CORRELATED, + + /** + * Publisher confirms are disabled (default). + */ + NONE + } private final Set allocatedConnections = new HashSet<>(); @@ -176,9 +203,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory private int connectionLimit = Integer.MAX_VALUE; - private boolean publisherConfirms; - - private boolean simplePublisherConfirms; + private ConfirmType confirmType = ConfirmType.NONE; private boolean publisherReturns; @@ -356,7 +381,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory @Override public boolean isPublisherConfirms() { - return this.publisherConfirms; + return ConfirmType.CORRELATED.equals(this.confirmType); } @Override @@ -372,36 +397,50 @@ public class CachingConnectionFactory extends AbstractConnectionFactory } /** - * Use full publisher confirms, with correlation data and a callback for each message. + * Use full (correlated) publisher confirms, with correlation data and a callback for + * each message. * @param publisherConfirms true for full publisher returns, * @since 1.1 + * @deprecated in favor of {@link #setPublisherConfirmType(ConfirmType)}. * @see #setSimplePublisherConfirms(boolean) */ + @Deprecated public void setPublisherConfirms(boolean publisherConfirms) { - Assert.isTrue(!this.simplePublisherConfirms, "Cannot set both publisherConfirms and simplePublisherConfirms"); - this.publisherConfirms = publisherConfirms; - if (this.publisherConnectionFactory != null) { - this.publisherConnectionFactory.setPublisherConfirms(publisherConfirms); - } + Assert.isTrue(!ConfirmType.SIMPLE.equals(this.confirmType), + "Cannot set both publisherConfirms and simplePublisherConfirms"); + setPublisherConfirmType(ConfirmType.CORRELATED); } /** * Use simple publisher confirms where the template simply waits for completion. * @param simplePublisherConfirms true for confirms. * @since 2.1 + * @deprecated in favor of {@link #setPublisherConfirmType(ConfirmType)}. * @see #setPublisherConfirms(boolean) */ + @Deprecated public void setSimplePublisherConfirms(boolean simplePublisherConfirms) { - Assert.isTrue(!this.publisherConfirms, "Cannot set both publisherConfirms and simplePublisherConfirms"); - this.simplePublisherConfirms = simplePublisherConfirms; - if (this.publisherConnectionFactory != null) { - this.publisherConnectionFactory.setSimplePublisherConfirms(simplePublisherConfirms); - } + Assert.isTrue(!ConfirmType.CORRELATED.equals(this.confirmType), + "Cannot set both publisherConfirms and simplePublisherConfirms"); + setPublisherConfirmType(ConfirmType.SIMPLE); } @Override public boolean isSimplePublisherConfirms() { - return this.simplePublisherConfirms; + return this.confirmType.equals(ConfirmType.SIMPLE); + } + + /** + * Set the confirm type to use; default {@link ConfirmType#NONE}. + * @param confirmType the confirm type. + * @since 2.2 + */ + public void setPublisherConfirmType(ConfirmType confirmType) { + Assert.notNull(confirmType, "'confirmType' cannot be null"); + this.confirmType = confirmType; + if (this.publisherConnectionFactory != null) { + this.publisherConnectionFactory.setPublisherConfirmType(confirmType); + } } /** @@ -630,7 +669,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory } getChannelListener().onCreate(targetChannel, transactional); Class[] interfaces; - if (this.publisherConfirms || this.publisherReturns) { + if (ConfirmType.CORRELATED.equals(this.confirmType) || this.publisherReturns) { interfaces = new Class[] { ChannelProxy.class, PublisherCallbackChannel.class }; } else { @@ -672,7 +711,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory private Channel doCreateBareChannel(ChannelCachingConnectionProxy conn, boolean transactional) { Channel channel = conn.createBareChannel(transactional); - if (this.publisherConfirms || this.simplePublisherConfirms) { + if (!ConfirmType.NONE.equals(this.confirmType)) { try { channel.confirmSelect(); } @@ -680,7 +719,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory logger.error("Could not configure the channel to receive publisher confirms", e); } } - if ((this.publisherConfirms || this.publisherReturns) + if ((ConfirmType.CORRELATED.equals(this.confirmType) || this.publisherReturns) && !(channel instanceof PublisherCallbackChannelImpl)) { channel = this.publisherChannelFactory.createChannel(channel, getChannelsExecutor()); } @@ -1037,9 +1076,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory private final boolean transactional; - private final boolean confirmSelected = CachingConnectionFactory.this.simplePublisherConfirms; + private final boolean confirmSelected = ConfirmType.SIMPLE.equals(CachingConnectionFactory.this.confirmType); - private final boolean publisherConfirms = CachingConnectionFactory.this.publisherConfirms; + private final boolean publisherConfirms = + ConfirmType.CORRELATED.equals(CachingConnectionFactory.this.confirmType); private volatile Channel target; @@ -1285,7 +1325,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory boolean async = false; try { if (CachingConnectionFactory.this.active && - (CachingConnectionFactory.this.publisherConfirms || + (ConfirmType.CORRELATED.equals(CachingConnectionFactory.this.confirmType) || CachingConnectionFactory.this.publisherReturns)) { async = true; asyncClose(proxy); @@ -1317,7 +1357,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory try { executorService.execute(() -> { try { - if (CachingConnectionFactory.this.publisherConfirms) { + if (ConfirmType.CORRELATED.equals(CachingConnectionFactory.this.confirmType)) { channel.waitForConfirmsOrDie(ASYNC_CLOSE_TIMEOUT); } else { diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd index 01b1fbeb..4f35c551 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd @@ -1454,10 +1454,21 @@ + + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java index 9441cd4f..d2eba7cf 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java @@ -42,6 +42,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -463,7 +464,7 @@ public class AsyncRabbitTemplateTests { @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost"); - connectionFactory.setPublisherConfirms(true); + connectionFactory.setPublisherConfirmType(ConfirmType.CORRELATED); connectionFactory.setPublisherReturns(true); return connectionFactory; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java index 77e75445..39c74b7f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java @@ -25,6 +25,7 @@ import org.junit.Before; import org.junit.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; @@ -61,7 +62,7 @@ public final class ConnectionFactoryParserTests { assertThat(connectionFactory.getChannelCacheSize()).isEqualTo(10); DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory); assertThat(dfa.getPropertyValue("executorService")).isNull(); - assertThat(dfa.getPropertyValue("publisherConfirms")).isEqualTo(Boolean.TRUE); + assertThat(dfa.getPropertyValue("confirmType")).isEqualTo(ConfirmType.CORRELATED); assertThat(dfa.getPropertyValue("publisherReturns")).isEqualTo(Boolean.TRUE); assertThat(TestUtils.getPropertyValue(connectionFactory, "rabbitConnectionFactory.requestedHeartbeat")).isEqualTo(123); assertThat(TestUtils.getPropertyValue(connectionFactory, "rabbitConnectionFactory.connectionTimeout")).isEqualTo(789); @@ -87,7 +88,7 @@ public final class ConnectionFactoryParserTests { ThreadPoolTaskExecutor exec = beanFactory.getBean("exec", ThreadPoolTaskExecutor.class); assertThat(executor).isSameAs(exec.getThreadPoolExecutor()); DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory); - assertThat(dfa.getPropertyValue("publisherConfirms")).isEqualTo(Boolean.FALSE); + assertThat(dfa.getPropertyValue("confirmType")).isEqualTo(ConfirmType.NONE); assertThat(dfa.getPropertyValue("publisherReturns")).isEqualTo(Boolean.FALSE); assertThat(connectionFactory.getCacheMode()).isEqualTo(CachingConnectionFactory.CacheMode.CONNECTION); assertThat(TestUtils.getPropertyValue(connectionFactory, "rabbitConnectionFactory.connectionTimeout")).isEqualTo(new ConnectionFactory().getConnectionTimeout()); @@ -103,6 +104,8 @@ public final class ConnectionFactoryParserTests { assertThat(executor).isNotNull(); ExecutorService exec = beanFactory.getBean("execService", ExecutorService.class); assertThat(executor).isSameAs(exec); + DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory); + assertThat(dfa.getPropertyValue("confirmType")).isEqualTo(ConfirmType.SIMPLE); } @Test diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index e67e3c45..2275d419 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java @@ -73,6 +73,7 @@ import org.mockito.InOrder; import org.springframework.amqp.AmqpConnectException; import org.springframework.amqp.AmqpTimeoutException; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.context.ApplicationContext; @@ -623,7 +624,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest ccf.setExecutor(exec); ccf.setChannelCacheSize(1); ccf.setChannelCheckoutTimeout(1); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final Connection con = ccf.createConnection(); @@ -692,7 +693,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest ccf.setExecutor(mock(ExecutorService.class)); ccf.setChannelCacheSize(1); ccf.setChannelCheckoutTimeout(1); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); RabbitTemplate rabbitTemplate = new RabbitTemplate(ccf); rabbitTemplate.convertAndSend("foo", "bar"); @@ -1579,7 +1580,9 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(executor); - ccf.setPublisherConfirms(confirms); + if (confirms) { + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); + } Connection con = ccf.createConnection(); Channel channel = con.createChannel(false); @@ -1746,7 +1749,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest given(mockConnection.isOpen()).willReturn(true); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ApplicationContext ac = mock(ApplicationContext.class); ccf.setApplicationContext(ac); PublisherCallbackChannel pcc = mock(PublisherCallbackChannel.class); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 601bafdf..155b66b0 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -72,6 +72,7 @@ import org.springframework.amqp.core.ReceiveAndReplyCallback; import org.springframework.amqp.core.ReceiveAndReplyMessageCallback; import org.springframework.amqp.core.ReplyToAddressCallback; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ChannelListener; import org.springframework.amqp.rabbit.connection.ClosingRecoveryListener; import org.springframework.amqp.rabbit.connection.Connection; @@ -1601,7 +1602,7 @@ public class RabbitTemplateIntegrationTests { @Test public void waitForConfirms() { - this.connectionFactory.setPublisherConfirms(true); + this.connectionFactory.setPublisherConfirmType(ConfirmType.CORRELATED); Collection messages = getMessagesToSend(); Boolean result = this.template.invoke(t -> { messages.forEach(m -> t.convertAndSend(ROUTE, m)); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 3a117509..60371132 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -60,6 +60,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.core.QueueBuilder.Overflow; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ChannelProxy; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.connection.PendingConfirm; @@ -121,7 +122,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { connectionFactoryWithConfirmsEnabled.setHost("localhost"); connectionFactoryWithConfirmsEnabled.setChannelCacheSize(100); connectionFactoryWithConfirmsEnabled.setPort(BrokerTestUtils.getPort()); - connectionFactoryWithConfirmsEnabled.setPublisherConfirms(true); + connectionFactoryWithConfirmsEnabled.setPublisherConfirmType(ConfirmType.CORRELATED); templateWithConfirmsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsEnabled); connectionFactoryWithReturnsEnabled = new CachingConnectionFactory(); connectionFactoryWithReturnsEnabled.setHost("localhost"); @@ -133,7 +134,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { connectionFactoryWithConfirmsAndReturnsEnabled.setHost("localhost"); connectionFactoryWithConfirmsAndReturnsEnabled.setChannelCacheSize(100); connectionFactoryWithConfirmsAndReturnsEnabled.setPort(BrokerTestUtils.getPort()); - connectionFactoryWithConfirmsAndReturnsEnabled.setPublisherConfirms(true); + connectionFactoryWithConfirmsAndReturnsEnabled.setPublisherConfirmType(ConfirmType.CORRELATED); connectionFactoryWithConfirmsAndReturnsEnabled.setPublisherReturns(true); templateWithConfirmsAndReturnsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsAndReturnsEnabled); templateWithConfirmsAndReturnsEnabled.setMandatory(true); @@ -331,7 +332,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(ccf); final AtomicBoolean confirmed = new AtomicBoolean(); @@ -366,7 +367,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.setChannelCacheSize(3); final RabbitTemplate template = new RabbitTemplate(ccf); @@ -440,7 +441,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(ccf); final AtomicBoolean confirmed = new AtomicBoolean(); @@ -482,7 +483,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(ccf); final CountDownLatch latch = new CountDownLatch(2); @@ -522,7 +523,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template1 = new RabbitTemplate(ccf); final Set confirms = new HashSet(); @@ -581,7 +582,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.setChannelCacheSize(3); final RabbitTemplate template = new RabbitTemplate(ccf); @@ -704,7 +705,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(mock(ExecutorService.class)); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(ccf); final AtomicBoolean confirmed = new AtomicBoolean(); @@ -766,7 +767,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setExecutor(Executors.newSingleThreadExecutor()); - ccf.setPublisherConfirms(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(ccf); final CountDownLatch confirmed = new CountDownLatch(1); 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 cffa1f1b..7b48f2dd 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 @@ -26,6 +26,7 @@ import org.junit.jupiter.api.BeforeEach; 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.junit.BrokerTestUtils; import org.springframework.amqp.rabbit.junit.RabbitAvailable; @@ -55,7 +56,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 { // otherwise channels can be closed before confirms are received. connectionFactoryWithConfirmsEnabled.setChannelCacheSize(100); connectionFactoryWithConfirmsEnabled.setPort(BrokerTestUtils.getPort()); - connectionFactoryWithConfirmsEnabled.setPublisherConfirms(true); + connectionFactoryWithConfirmsEnabled.setPublisherConfirmType(ConfirmType.CORRELATED); templateWithConfirmsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsEnabled); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java index 836c8727..fe29f4bf 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java @@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger; 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.Connection; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.junit.RabbitAvailable; @@ -55,7 +56,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { public void testRepublishOnNackThreadNoExchange() throws Exception { CachingConnectionFactory cf = new CachingConnectionFactory( RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); - cf.setPublisherConfirms(true); + cf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(cf); final CountDownLatch confirmLatch = new CountDownLatch(2); template.setConfirmCallback((cd, a, c) -> { @@ -74,7 +75,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { final CachingConnectionFactory cf = new CachingConnectionFactory( RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); cf.setPublisherReturns(true); - cf.setPublisherConfirms(true); + cf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(cf); final CountDownLatch returnLatch = new CountDownLatch(1); final CountDownLatch confirmLatch = new CountDownLatch(1); @@ -108,7 +109,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { public void testDeferredChannelCacheAck() throws Exception { final CachingConnectionFactory cf = new CachingConnectionFactory( RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); - cf.setPublisherConfirms(true); + cf.setPublisherConfirmType(ConfirmType.CORRELATED); final RabbitTemplate template = new RabbitTemplate(cf); final CountDownLatch confirmLatch = new CountDownLatch(1); final AtomicInteger cacheCount = new AtomicInteger(); @@ -134,7 +135,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { public void testTwoSendsAndReceivesDRTMLC() throws Exception { CachingConnectionFactory cf = new CachingConnectionFactory( RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); - cf.setPublisherConfirms(true); + cf.setPublisherConfirmType(ConfirmType.CORRELATED); RabbitTemplate template = new RabbitTemplate(cf); template.setReplyTimeout(0); final CountDownLatch confirmLatch = new CountDownLatch(2); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/SimplePublisherConfirmsTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/SimplePublisherConfirmsTests.java index ca0b1b60..3f5714b5 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/SimplePublisherConfirmsTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/SimplePublisherConfirmsTests.java @@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.junit.RabbitAvailable; /** @@ -41,7 +42,7 @@ public class SimplePublisherConfirmsTests { @Test public void testConfirms() { CachingConnectionFactory cf = new CachingConnectionFactory("localhost"); - cf.setSimplePublisherConfirms(true); + cf.setPublisherConfirmType(ConfirmType.SIMPLE); RabbitTemplate template = new RabbitTemplate(cf); template.setRoutingKey(QUEUE); Boolean invokeResult = template.invoke(t -> { @@ -57,7 +58,7 @@ public class SimplePublisherConfirmsTests { @Test public void testConfirmsWithCallbacks() { CachingConnectionFactory cf = new CachingConnectionFactory("localhost"); - cf.setSimplePublisherConfirms(true); + cf.setPublisherConfirmType(ConfirmType.SIMPLE); RabbitTemplate template = new RabbitTemplate(cf); template.setRoutingKey(QUEUE); AtomicReference finalProperties = new AtomicReference<>(); diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml index ac0f124d..90ef5660 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml @@ -31,6 +31,7 @@ diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 8fa52b05..8af7732b 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -685,7 +685,7 @@ mastered and connects to the address in the same array position. [[cf-pub-conf-ret]] ===== Publisher Confirms and Returns -Confirmed and returned messages are supported by setting the `publisherConfirms` and `publisherReturns` properties of the `CachingConnectionFactory` to 'true'. +Confirmed (with correlation) and returned messages are supported by setting the `CachingConnectionFactory` property `publisherConfirmType` to `ConfirmType.CORRELATED` and the `publisherReturns` property to 'true'. When these options are set, `Channel` instances created by the factory are wrapped in an `PublisherCallbackChannel`, which is used to facilitate the callbacks. When such a channel is obtained, the client can register a `PublisherCallbackChannel.Listener` with the `Channel`. @@ -1059,7 +1059,7 @@ void returnedMessage(Message message, int replyCode, String replyText, Only one `ReturnCallback` is supported by each `RabbitTemplate`. See also <>. -For publisher confirms (also known as publisher acknowledgements), the template requires a `CachingConnectionFactory` that has its `publisherConfirms` property set to `true`. +For publisher confirms (also known as publisher acknowledgements), the template requires a `CachingConnectionFactory` that has its `publisherConfirm` property set to `ConfirmType.CORRELATED`. Confirms are sent to the client by it registering a `RabbitTemplate.ConfirmCallback` by calling `setConfirmCallback(ConfirmCallback callback)`. The callback must implement this method: @@ -1105,7 +1105,7 @@ This is no longer necessary since the framework now hands off the callback invoc IMPORTANT: The guarantee of receiving a returned message before the ack is still maintained as long as the return callback executes in 60 seconds or less. The confirm is scheduled to be delivered after the return callback exits or after 60 seconds, whichever comes first. -Starting with version 2.1, the `CorrelationData` object has a `ListenableFuture` that you can \used to get the result, instead of using a `ConfirmCallback` on the template. +Starting with version 2.1, the `CorrelationData` object has a `ListenableFuture` that you can use to get the result, instead of using a `ConfirmCallback` on the template. The following example shows how to configure a `CorrelationData` instance: ==== @@ -1176,9 +1176,9 @@ NOTE: The preceding discussion is moot if the template operations are already pe In that case, the operations are performed on that channel and committed when the thread returns to the container. It is not necessary to use `invoke` in that scenario. -When using confirms in this way, much of the infrastructure set up for correlating confirms to requests is not really needed. -Starting with version 2.1, the connection factory supports a new property called `simplePublisherConfirms`. -When this is `true`, the infrastructure is avoided and the confirm processing can be more efficient. +When using confirms in this way, much of the infrastructure set up for correlating confirms to requests is not really needed (unless returns are also enabled). +Starting with version 2.2, the connection factory supports a new property called `publisherConfirmType`. +When this is set to `ConfirmType.SIMPLE`, the infrastructure is avoided and the confirm processing can be more efficient. Furthermore, the `RabbitTemplate` sets the `publisherSequenceNumber` property in the sent message `MessageProperties`. If you wish to check (or log or otherwise use) specific confirms, you can do so with an overloaded `invoke` method, as the following example shows: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index cd53390a..14a17a1a 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -79,6 +79,8 @@ When using Publisher confirms and returns, the callbacks are now invoked on the This avoids a possible deadlock in the `amqp-clients` library if you perform rabbit operations from within the callback. See <> for more information. +Also, the publisher confirm type is now specified with the `ConfirmType` enum instead of the two mutually exclusive setter methods. + ===== Other Changes The `Declarables` object (for declaring multiple queues, exchanges, bindings) now has a filtered getter for each type.