SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-121: Add Pollable Consumer

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/121
Resolves #125
This commit is contained in:
Gary Russell
2018-01-10 17:19:26 -05:00
committed by Oleg Zhurakousky
parent 100f7749ce
commit 004b3eafba
4 changed files with 200 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-cloud-stream-binder-rabbit/pom.xml
View File

@@ -51,14 +51,17 @@
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-amqp</artifactId>
<version>5.0.1.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>5.0.1.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jmx</artifactId>
<version>5.0.1.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>

68
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2017 the original author or authors.
* Copyright 2013-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -47,6 +47,7 @@ import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
@@ -60,14 +61,18 @@ import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.inbound.AmqpMessageSource;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.support.AcknowledgmentCallback;
import org.springframework.integration.support.AcknowledgmentCallback.Status;
import org.springframework.integration.support.DefaultErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
@@ -306,7 +311,7 @@ public class RabbitMessageChannelBinder
}
}
public boolean expressionInterceptorNeeded(RabbitProducerProperties extendedProperties) {
private boolean expressionInterceptorNeeded(RabbitProducerProperties extendedProperties) {
return extendedProperties.getRoutingKeyExpression() != null
&& extendedProperties.getRoutingKeyExpression().contains("payload")
|| (extendedProperties.getDelayExpression() != null
@@ -363,7 +368,7 @@ public class RabbitMessageChannelBinder
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
listenerContainer.setQueueNames(consumerDestination.getName());
listenerContainer.setQueueNames(destination);
listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
listenerContainer.setMessagePropertiesConverter(
RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
@@ -398,6 +403,25 @@ public class RabbitMessageChannelBinder
return adapter;
}
@Override
protected PolledConsumerResources createPolledConsumerResources(String name, String group, ConsumerDestination destination,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
AmqpMessageSource source = new AmqpMessageSource(this.connectionFactory, destination.getName());
source.setRawMessageHeader(true);
return new PolledConsumerResources(source,
registerErrorInfrastructure(destination, group, consumerProperties, true));
}
@Override
protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget) {
bindingTarget.setAttributesProvider((accessor, message) -> {
Object rawMessage = message.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (rawMessage != null) {
accessor.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, rawMessage);
}
});
}
@Override
protected ErrorMessageStrategy getErrorMessageStrategy() {
return errorMessageStrategy;
@@ -481,6 +505,44 @@ public class RabbitMessageChannelBinder
}
}
@Override
protected MessageHandler getPolledConsumerErrorMessageHandler(ConsumerDestination destination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
MessageHandler handler = getErrorMessageHandler(destination, group, properties);
if (handler != null) {
return handler;
}
final MessageHandler superHandler = super.getErrorMessageHandler(destination, group, properties);
return message -> {
Message amqpMessage = (Message) message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (!(message instanceof ErrorMessage)) {
logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: "
+ message);
}
else if (amqpMessage == null) {
if (superHandler != null) {
superHandler.handleMessage(message);
}
}
else {
if (message.getPayload() instanceof MessagingException) {
AcknowledgmentCallback ack = StaticMessageHeaderAccessor.getAcknowledgmentCallback(
((MessagingException) message.getPayload()).getFailedMessage());
if (ack != null) {
if (properties.getExtension().isRequeueRejected()) {
ack.acknowledge(Status.REQUEUE);
}
else {
ack.acknowledge(Status.REJECT);
}
}
}
}
};
}
@Override
protected String errorsBaseName(ConsumerDestination destination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {

111
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2017 the original author or authors.
* Copyright 2013-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -60,12 +60,14 @@ import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.stream.binder.BinderException;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
import org.springframework.cloud.stream.binder.PartitionTestSupport;
import org.springframework.cloud.stream.binder.PollableSource;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
@@ -87,6 +89,7 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
@@ -1324,6 +1327,112 @@ public class RabbitBinderTests extends
producerBinding.unbind();
}
@Test
public void testPolledConsumer() throws Exception {
RabbitTestBinder binder = getBinder();
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource();
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollable", "group",
inboundBindTarget, createConsumerProperties());
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("pollable.group", "testPollable");
boolean polled = inboundBindTarget.poll(m -> {
assertThat(m.getPayload()).isEqualTo("testPollable");
});
int n = 0;
while (n++ < 100 && !polled) {
polled = inboundBindTarget.poll(m -> {
assertThat(m.getPayload()).isEqualTo("testPollable");
});
}
assertThat(polled).isTrue();
binding.unbind();
}
@Test
public void testPolledConsumerWithDlq() throws Exception {
RabbitTestBinder binder = getBinder();
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.setMaxAttempts(2);
properties.setBackOffInitialInterval(0);
properties.getExtension().setAutoBindDlq(true);
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollableDlq", "group",
inboundBindTarget, properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("pollableDlq.group", "testPollable");
try {
int n = 0;
while (n++ < 100) {
inboundBindTarget.poll(m -> {
throw new RuntimeException("test DLQ");
});
Thread.sleep(100);
}
}
catch (MessageHandlingException e) {
assertThat(e.getCause().getCause().getCause().getCause().getCause().getMessage()).isEqualTo("test DLQ");
}
org.springframework.amqp.core.Message deadLetter = template.receive("pollableDlq.group.dlq", 10_000);
assertThat(deadLetter).isNotNull();
binding.unbind();
}
@Test
public void testPolledConsumerWithDlqNoRetry() throws Exception {
RabbitTestBinder binder = getBinder();
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.setMaxAttempts(1);
// properties.getExtension().setRequeueRejected(true); // loops, correctly
properties.getExtension().setAutoBindDlq(true);
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollableDlqNoRetry", "group",
inboundBindTarget, properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("pollableDlqNoRetry.group", "testPollable");
try {
int n = 0;
while (n++ < 100) {
inboundBindTarget.poll(m -> {
throw new RuntimeException("test DLQ");
});
Thread.sleep(100);
}
}
catch (MessageHandlingException e) {
assertThat(e.getCause().getMessage()).isEqualTo("test DLQ");
}
org.springframework.amqp.core.Message deadLetter = template.receive("pollableDlqNoRetry.group.dlq", 10_000);
assertThat(deadLetter).isNotNull();
binding.unbind();
}
@Test
public void testPolledConsumerWithDlqRePub() throws Exception {
RabbitTestBinder binder = getBinder();
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.setMaxAttempts(2);
properties.setBackOffInitialInterval(0);
properties.getExtension().setAutoBindDlq(true);
properties.getExtension().setRepublishToDlq(true);
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollableDlqRePub", "group",
inboundBindTarget, properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("pollableDlqRePub.group", "testPollable");
boolean polled = false;
int n = 0;
while (n++ < 100 && !polled) {
Thread.sleep(100);
polled = inboundBindTarget.poll(m -> {
throw new RuntimeException("test DLQ");
});
}
assertThat(polled).isTrue();
org.springframework.amqp.core.Message deadLetter = template.receive("pollableDlqRePub.group.dlq", 10_000);
assertThat(deadLetter).isNotNull();
binding.unbind();
}
private SimpleMessageListenerContainer verifyContainer(Lifecycle endpoint) {
SimpleMessageListenerContainer container;
RetryTemplate retry;

27
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 the original author or authors.
* Copyright 2015-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,10 +22,11 @@ import java.util.Set;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.cloud.stream.binder.AbstractPollableConsumerTestBinder;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.PollableSource;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
@@ -34,6 +35,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
/**
* Test support class for {@link RabbitMessageChannelBinder}.
@@ -43,7 +45,9 @@ import org.springframework.messaging.MessageChannel;
* @author David Turanski
* @author Mark Fisher
*/
public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
public class RabbitTestBinder extends
AbstractPollableConsumerTestBinder<RabbitMessageChannelBinder,
ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private final RabbitAdmin rabbitAdmin;
@@ -64,7 +68,7 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
this.applicationContext = new AnnotationConfigApplicationContext(Config.class);
binder.setApplicationContext(this.applicationContext);
binder.setProducerConnectionFactory(connectionFactory);
this.setBinder(binder);
this.setPollableConsumerBinder(binder);
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
}
@@ -75,6 +79,20 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
captureConsumerResources(name, group, properties);
return super.bindConsumer(name, group, moduleInputChannel, properties);
}
@Override
public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name, String group,
PollableSource<MessageHandler> inboundBindTarget,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
captureConsumerResources(name, group, properties);
return super.bindPollableConsumer(name, group, inboundBindTarget, properties);
}
private void captureConsumerResources(String name, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
if (group != null) {
if (properties.getExtension().isQueueNameGroupOnly()) {
this.queues.add(properties.getExtension().getPrefix() + group);
@@ -86,7 +104,6 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
this.exchanges.add(properties.getExtension().getPrefix() + name);
this.prefixes.add(properties.getExtension().getPrefix());
deadLetters(properties.getExtension());
return super.bindConsumer(name, group, moduleInputChannel, properties);
}
@Override