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

Fix AMQP MessageSource Tests

Browse Source 
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
This commit is contained in:
Gary Russell
2018-03-22 16:31:27 -04:00
committed by Artem Bilan
parent cc63104f11
commit a5c16fa0ac
2 changed files with 37 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java
View File

@@ -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<Object> {
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<Object> {
@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<Object> {
}
}
@Override
public String toString() {
return "AmqpAckCallback [ackInfo=" + this.ackInfo + ", acknowledged=" + this.acknowledged
+ ", autoAckEnabled=" + this.autoAckEnabled + "]";
}
}
/**
@@ -302,6 +316,12 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
return this.getResponse;
}
@Override
public String toString() {
return "AmqpAckInfo [connection=" + this.connection + ", channel=" + this.channel + ", transacted="
+ this.transacted + ", getResponse=" + this.getResponse + "]";
}
}
}

19
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java
View File

@@ -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<Object> doReceive() {
AbstractIntegrationMessageBuilder<Object> builder = super.doReceive();
if (builder != null) {
Config.this.requeueLatch.countDown();
}
return builder;
}
};
}
@ServiceActivator(inputChannel = "noAutoAck")