diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 35626de0..6c40d2f8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -175,6 +175,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } } + protected ExecutorService getExecutorService() { + return executorService; + } + /** * How long to wait (milliseconds) for a response to a connection close * operation from the broker; default 30000 (30 seconds). @@ -185,6 +189,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di this.closeTimeout = closeTimeout; } + public int getCloseTimeout() { + return closeTimeout; + } + protected final Connection createBareConnection() { try { if (this.addresses != null) { 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 12b28f63..3ebaf33f 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 @@ -25,6 +25,8 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; @@ -115,6 +117,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); + /** Executor used for deferred close if no explicit executor set. */ + private final ExecutorService deferredCloseExecutor = Executors.newCachedThreadPool(); + + /** * Create a new CachingConnectionFactory initializing the hostname to be the value returned from * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception. @@ -612,7 +618,45 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen return; } try { - this.target.close(); + if (CachingConnectionFactory.this.active && + (CachingConnectionFactory.this.publisherConfirms || + CachingConnectionFactory.this.publisherReturns)) { + ExecutorService executorService = (getExecutorService() != null + ? getExecutorService() + : CachingConnectionFactory.this.deferredCloseExecutor); + final Channel channel = CachedChannelInvocationHandler.this.target; + executorService.execute(new Runnable() { + + @Override + public void run() { + try { + if (CachingConnectionFactory.this.publisherConfirms) { + channel.waitForConfirmsOrDie(5000); + } + else { + Thread.sleep(5000); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + catch (Exception e) {} + finally { + try { + if (channel.isOpen()) { + channel.close(); + } + } + catch (IOException e) {} + catch (AlreadyClosedException e) {} + } + } + + }); + } + else { + this.target.close(); + } } catch (AlreadyClosedException e) { if (logger.isTraceEnabled()) { @@ -642,8 +686,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen @Override public Channel createChannel(boolean transactional) { - Channel channel = getChannel(this, transactional); - return channel; + return getChannel(this, transactional); } @Override diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java index 913b01f5..2f86916a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java @@ -73,15 +73,19 @@ import com.rabbitmq.client.ShutdownSignalException; * @since 1.0.1 * */ -public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener { +public class PublisherCallbackChannelImpl + implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener { - private static final String[] METHODS_OF_INTEREST = new String[] {"getFlow", "flow", "flowBlocked", "basicConsume", "basicQos"}; + private static final String[] METHODS_OF_INTEREST = + new String[] {"getFlow", "flow", "flowBlocked", "basicConsume", "basicQos"}; private static final MethodFilter METHOD_FILTER = new MethodFilter() { + @Override public boolean matches(java.lang.reflect.Method method) { return ObjectUtils.containsElement(METHODS_OF_INTEREST, method.getName()); } + }; private final Log logger = LogFactory.getLog(this.getClass()); @@ -109,7 +113,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C delegate.addShutdownListener(this); this.delegate = delegate; - // The following reflection is required to maintain comatibility with pre 3.3.x clients. + // The following reflection is required to maintain compatibility with pre 3.3.x clients. final AtomicReference getFlowMethod = new AtomicReference(); final AtomicReference flowMethod = new AtomicReference(); final AtomicReference flowBlockedMethod = new AtomicReference(); @@ -454,7 +458,8 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C return (String) ReflectionUtils.invokeMethod(this.basicConsumeFourArgsMethod, this.delegate, queue, autoAck, arguments, callback); } - throw new UnsupportedOperationException("'basicConsume(String, boolean, Map, Consumer)' is not supported by the client library"); + throw new UnsupportedOperationException("'basicConsume(String, boolean, Map, Consumer)' " + + "is not supported by the client library"); } public String basicConsume(String queue, boolean autoAck, @@ -719,7 +724,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C AMQP.BasicProperties properties, byte[] body) throws IOException { - Object uuidObject = properties.getHeaders().get(RETURN_CORRELATION).toString(); + String uuidObject = properties.getHeaders().get(RETURN_CORRELATION).toString(); Listener listener = this.listeners.get(uuidObject); if (listener == null || !listener.isReturnListener()) { if (logger.isWarnEnabled()) { @@ -748,10 +753,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C @Override public boolean equals(Object obj) { - if (obj == this) { - return true; - } - return this.delegate.equals(obj); + return obj == this || this.delegate.equals(obj); } @Override diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java index 2b039704..d63a0ccc 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java @@ -88,7 +88,7 @@ public class RabbitTemplatePerformanceIntegrationTests { } @Test - @Repeat(2000) + @Repeat(200) public void testSendAndReceive() throws Exception { template.convertAndSend(ROUTE, "message"); String result = (String) template.receiveAndConvert(ROUTE); @@ -104,7 +104,7 @@ public class RabbitTemplatePerformanceIntegrationTests { } @Test - @Repeat(2000) + @Repeat(200) public void testSendAndReceiveTransacted() throws Exception { template.setChannelTransacted(true); template.convertAndSend(ROUTE, "message"); @@ -113,7 +113,7 @@ public class RabbitTemplatePerformanceIntegrationTests { } @Test - @Repeat(2000) + @Repeat(200) public void testSendAndReceiveExternalTransacted() throws Exception { template.setChannelTransacted(true); new TransactionTemplate(new TestTransactionManager()).execute(new TransactionCallback() { 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 16bde278..2a78daca 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 @@ -74,11 +74,11 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; -import com.rabbitmq.client.Consumer; /** * @author Gary Russell * @author Gunar Hillert + * @author Artem Bilan * @since 1.1 * */ @@ -460,14 +460,12 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ccf.setPublisherConfirms(true); final RabbitTemplate template = new RabbitTemplate(ccf); - final List confirms = new ArrayList(); final CountDownLatch latch = new CountDownLatch(2); template.setConfirmCallback(new ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack) { - confirms.add(correlationData.getId()); latch.countDown(); } } @@ -628,8 +626,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { // 3.3.1 client - channel.basicConsume("foo", false, (Map) null, (Consumer) null); - verify(mockChannel).basicConsume("foo", false, (Map) null, (Consumer) null); + channel.basicConsume("foo", false, (Map) null, null); + verify(mockChannel).basicConsume("foo", false, (Map) null, null); channel.basicQos(3, false); verify(mockChannel).basicQos(3, false); @@ -717,4 +715,63 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { assertThat(log.get(), containsString("NOT_FOUND - no exchange '" + exchange)); } + @Test + public void testConfirmReceivedAfterPublisherCallbackChannelScheduleClose() throws Exception { + final CountDownLatch latch = new CountDownLatch(40); + templateWithConfirmsEnabled.setConfirmCallback(new ConfirmCallback() { + + @Override + public void confirm(CorrelationData correlationData, boolean ack, String cause) { + latch.countDown(); + } + }); + + ExecutorService executorService = Executors.newCachedThreadPool(); + for (int i = 0; i < 20; i++) { + executorService.execute(new Runnable() { + + @Override + public void run() { + templateWithConfirmsEnabled.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + templateWithConfirmsEnabled.convertAndSend("BAD_ROUTE", (Object) "bad", new CorrelationData("cba")); + } + + }); + } + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertNull(templateWithConfirmsEnabled.getUnconfirmed(0)); + } + + @Test + public void testReturnNotReceivedAfterPublisherCallbackChannelClose() throws Exception { + final CountDownLatch latch = new CountDownLatch(20); + templateWithReturnsEnabled.setMandatory(true); + templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() { + + @Override + public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { + latch.countDown(); + } + + }); + + ExecutorService executorService = Executors.newCachedThreadPool(); + for (int i = 0; i < 20; i++) { + executorService.execute(new Runnable() { + + @Override + public void run() { + templateWithReturnsEnabled.convertAndSend("BAD_ROUTE", (Object) "bad", new CorrelationData("cba")); + } + + }); + } + + executorService.shutdown(); + assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS)); + Thread.sleep(100); + assertFalse(latch.getCount() == 0); + } + } diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 81c9c58d..ec2a8dc9 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -563,12 +563,17 @@ public AmqpTemplate rabbitTemplate(); by calling setConfirmCallback(ConfirmCallback callback). The callback must implement this method: - - Publisher Confirms only work when the channel is cached. Otherwise, the channel is closed after the - publish operation so, by definition, cannot receive the confirmation. Be sure to set the + + When a rabbit template send operation completes, the channel is closed; this would preclude the reception + of confirms or returns in the case when the connection factory cache is full (when there is space in + the cache, the channel is not physically closed and the returns/confirms will proceed as normal). + When the cache is full, the framework defers the close for up to 5 seconds, in order to allow time + for the confirms/returns to be received. When using confirms, the channel will be closed when the + last confirm is received. When using only returns, the channel will remain open for the full + 5 seconds. It is generally recommended to set the connection factory's channelCacheSize to a large enough value so that the channel on which a - message is published is returned to the cache instead of being closed. - + message is published is returned to the cache instead of being closed. + The CorrelationData is an object supplied by the client when sending the