From a5c16fa0acc975fb96a5d9fc5fd1d135768854fc Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 22 Mar 2018 16:31:27 -0400 Subject: [PATCH] Fix AMQP MessageSource Tests https://build.spring.io/browse/INT-MASTER-983/ Race between the poller and template to get the requeued message. Also add trace logging to the acknowledgment. **cherry-pick to 5.0.x** # Conflicts: # spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java --- .../amqp/inbound/AmqpMessageSource.java | 20 +++++++++++++++++++ .../AmqpMessageSourceIntegrationTests.java | 19 ++++++++++++++++-- 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java index 5401652416..339305c8e2 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java @@ -20,6 +20,9 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -199,6 +202,8 @@ public class AmqpMessageSource extends AbstractMessageSource { public static class AmqpAckCallback implements AcknowledgmentCallback { + private static Log logger = LogFactory.getLog(AmqpAckCallback.class); + private final AmqpAckInfo ackInfo; private boolean acknowledged; @@ -235,6 +240,9 @@ public class AmqpMessageSource extends AbstractMessageSource { @Override public void acknowledge(Status status) { Assert.notNull(status, "'status' cannot be null"); + if (logger.isTraceEnabled()) { + logger.trace("acknowledge(" + status.name() + ") for " + this); + } try { long deliveryTag = this.ackInfo.getGetResponse().getEnvelope().getDeliveryTag(); switch (status) { @@ -264,6 +272,12 @@ public class AmqpMessageSource extends AbstractMessageSource { } } + @Override + public String toString() { + return "AmqpAckCallback [ackInfo=" + this.ackInfo + ", acknowledged=" + this.acknowledged + + ", autoAckEnabled=" + this.autoAckEnabled + "]"; + } + } /** @@ -302,6 +316,12 @@ public class AmqpMessageSource extends AbstractMessageSource { return this.getResponse; } + @Override + public String toString() { + return "AmqpAckInfo [connection=" + this.connection + ", channel=" + this.channel + ", transacted=" + + this.transacted + ", getResponse=" + this.getResponse + "]"; + } + } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java index 94d9c51f0e..55ef85a365 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java @@ -58,6 +58,7 @@ import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.Pollers; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.AcknowledgmentCallback; import org.springframework.integration.support.AcknowledgmentCallback.Status; import org.springframework.messaging.handler.annotation.Header; @@ -132,8 +133,9 @@ public class AmqpMessageSourceIntegrationTests { assertThat(this.config.fromDsl, equalTo("bar")); assertThat(this.config.fromInterceptedSource, equalTo("BAZ")); assertNull(template.receive(NOAUTOACK_QUEUE)); + assertThat(this.config.requeueLatch.getCount(), equalTo(1L)); this.config.callback.acknowledge(Status.REQUEUE); - assertNotNull(template.receive(NOAUTOACK_QUEUE, 10_000)); + assertTrue(this.config.requeueLatch.await(10, TimeUnit.SECONDS)); } @Configuration @@ -142,6 +144,8 @@ public class AmqpMessageSourceIntegrationTests { private final CountDownLatch latch = new CountDownLatch(5); + private final CountDownLatch requeueLatch = new CountDownLatch(2); + private volatile String received; private volatile Object fromDsl; @@ -174,7 +178,18 @@ public class AmqpMessageSourceIntegrationTests { @InboundChannelAdapter(channel = "noAutoAck", poller = @Poller(fixedDelay = "100"), autoStartup = "false") @Bean public MessageSource noAutoAckSource() { - return new AmqpMessageSource(connectionFactory(), NOAUTOACK_QUEUE); + return new AmqpMessageSource(connectionFactory(), NOAUTOACK_QUEUE) { + + @Override + protected AbstractIntegrationMessageBuilder doReceive() { + AbstractIntegrationMessageBuilder builder = super.doReceive(); + if (builder != null) { + Config.this.requeueLatch.countDown(); + } + return builder; + } + + }; } @ServiceActivator(inputChannel = "noAutoAck")