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 7e8b6a36..bec86945 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 @@ -623,7 +623,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory } if (this.publisherConfirms || this.publisherReturns) { if (!(channel instanceof PublisherCallbackChannelImpl)) { - channel = new PublisherCallbackChannelImpl(channel); + channel = new PublisherCallbackChannelImpl(channel, getExecutorService()); } } if (channel != null) { 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 b140b212..5c142778 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 @@ -31,6 +31,8 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeoutException; import org.apache.commons.logging.Log; @@ -79,6 +81,8 @@ import com.rabbitmq.client.ShutdownSignalException; public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener { + private static final ExecutorService DEFAULT_EXECUTOR = Executors.newSingleThreadExecutor(); + private final Log logger = LogFactory.getLog(this.getClass()); private final Channel delegate; @@ -91,9 +95,16 @@ public class PublisherCallbackChannelImpl private volatile java.util.function.Consumer afterAckCallback; + private final ExecutorService executor; + public PublisherCallbackChannelImpl(Channel delegate) { + this(delegate, null); + } + + public PublisherCallbackChannelImpl(Channel delegate, ExecutorService executor) { delegate.addShutdownListener(this); this.delegate = delegate; + this.executor = executor != null ? executor : DEFAULT_EXECUTOR; } @Override @@ -781,7 +792,11 @@ public class PublisherCallbackChannelImpl this.logger.trace(this.delegate + " is already closed"); } } - generateNacksForPendingAcks("Channel closed by application"); + shutdownCompleted("Channel closed by application"); + } + + private void shutdownCompleted(String cause) { + this.executor.execute(() -> generateNacksForPendingAcks(cause)); } private synchronized void generateNacksForPendingAcks(String cause) { @@ -997,7 +1012,7 @@ public class PublisherCallbackChannelImpl @Override public void shutdownCompleted(ShutdownSignalException cause) { - generateNacksForPendingAcks(cause.getMessage()); + shutdownCompleted(cause.getMessage()); } // Object 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 bd5f261e..5d5da6de 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 @@ -401,6 +401,10 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { exec.shutdown(); assertTrue(exec.awaitTermination(10, TimeUnit.SECONDS)); ccf.destroy(); + int n = 0; + while (n++ < 100 && pendingConfirms.size() > 0) { + Thread.sleep(100); + } assertEquals(0, pendingConfirms.size()); } @@ -739,7 +743,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { when(mockConnection.createChannel()).thenReturn(mockChannel1, mockChannel2); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); - ccf.setExecutor(mock(ExecutorService.class)); + ccf.setExecutor(Executors.newSingleThreadExecutor()); ccf.setPublisherConfirms(true); final RabbitTemplate template = new RabbitTemplate(ccf); @@ -768,13 +772,12 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { @Test public void testPublisherCallbackChannelImplCloseWithPending() throws Exception { - final AtomicInteger nacks = new AtomicInteger(); - Listener listener = mock(Listener.class); + final CountDownLatch latch = new CountDownLatch(2); doAnswer(invocation -> { boolean ack = invocation.getArgument(1); if (!ack) { - nacks.incrementAndGet(); + latch.countDown(); } return null; }).when(listener).handleConfirm(any(PendingConfirm.class), anyBoolean()); @@ -795,8 +798,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { channel.close(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + + int n = 0; + while (n++ < 100 && TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class).size() > 0) { + Thread.sleep(100); + } assertEquals(0, TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class).size()); - assertEquals(2, nacks.get()); } 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 f077c09e..3aea8909 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 @@ -23,7 +23,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -37,17 +36,18 @@ import com.rabbitmq.client.Channel; /** * @author Gary Russell - * * @since 2.1 * */ -@RabbitAvailable(queues = RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE) +@RabbitAvailable(queues = { RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE1, + RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE2 }) public class RabbitTemplatePublisherCallbacksIntegrationTests3 { - public static final String QUEUE = "defer.close"; + public static final String QUEUE1 = "synthetic.nack"; + + public static final String QUEUE2 = "defer.close"; @Test - @Disabled public void testRepublishOnNackThreadNoExchange() throws Exception { CachingConnectionFactory cf = new CachingConnectionFactory( RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); @@ -56,13 +56,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { final CountDownLatch confirmLatch = new CountDownLatch(2); template.setConfirmCallback((cd, a, c) -> { if (confirmLatch.getCount() == 2) { - template.convertAndSend(QUEUE, ((MyCD) cd).payload); // deadlock creating new channel + template.convertAndSend(QUEUE1, ((MyCD) cd).payload); } confirmLatch.countDown(); }); template.convertAndSend("bad.exchange", "junk", "foo", new MyCD("foo")); assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(template.receive(QUEUE, 10_000)).isNotNull(); + assertThat(template.receive(QUEUE1, 10_000)).isNotNull(); } @Test @@ -90,7 +90,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { channel2.close(); conn.close(); assertThat(TestUtils.getPropertyValue(cf, "cachedChannelsNonTransactional", List.class).size()).isEqualTo(2); - template.convertAndSend("", QUEUE + "junk", "foo", new MyCD("foo")); + template.convertAndSend("", QUEUE2 + "junk", "foo", new MyCD("foo")); assertThat(returnLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(cacheCount.get()).isEqualTo(1);