Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1067: Add enum for publisher confirm type

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1067

* * Refactor deprecated setters to use new setter

* * Fix this.publisherConnectionFactory typo
This commit is contained in:
Gary Russell
2019-08-19 11:21:21 -04:00
committed by Artem Bilan
parent bbc4ebabb4
commit acb56833d1
14 changed files with 130 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java
View File

@@ -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");
}
}

86
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -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<ChannelCachingConnectionProxy> 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 {

21
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd
View File

@@ -1454,10 +1454,21 @@
<xsd:attribute name="publisher-confirms" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true, channels on connections created by this factory support publisher confirms.
When true, channels on connections created by this factory support correlated publisher confirms.
DEPRECATED in favor of 'confirm-type'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-type" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The type of publisher confirmation to use, or NONE to disable.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="confirmTypes xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="publisher-returns" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -1594,6 +1605,14 @@
</xsd:restriction>
</xsd:simpleType>
<xsd:simpleType name="confirmTypes">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="SIMPLE"/>
<xsd:enumeration value="CORRELATED"/>
<xsd:enumeration value="NONE"/>
</xsd:restriction>
</xsd:simpleType>
<xsd:simpleType name="containerTypes">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="simple" />

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
View File

@@ -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;
}

7
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java
View File

@@ -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

11
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -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);

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -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));

21
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -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<String> confirms = new HashSet<String>();
@@ -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);

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

@@ -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);
}

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

@@ -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);

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/SimplePublisherConfirmsTests.java
View File

@@ -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<MessageProperties> finalProperties = new AtomicReference<>();

1
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml
View File

@@ -31,6 +31,7 @@
<rabbit:connection-factory id="withExecutorService" host="foo" virtual-host="/bar"
channel-cache-size="10" port="6888" username="user" password="password"
confirm-type="SIMPLE"
executor="execService" />
<bean id="execService" class="java.util.concurrent.Executors" factory-method="newSingleThreadExecutor" />

12
src/reference/asciidoc/amqp.adoc
View File

@@ -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 <<reply-timeout>>.
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:

2
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<template-confirms>> 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.