diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index 034201032..2ee395d03 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -51,14 +51,17 @@ org.springframework.integration spring-integration-amqp + 5.0.1.BUILD-SNAPSHOT org.springframework.integration spring-integration-core + 5.0.1.BUILD-SNAPSHOT org.springframework.integration spring-integration-jmx + 5.0.1.BUILD-SNAPSHOT org.springframework.integration diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index b8600844e..0eb4abc0c 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -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 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 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 consumerProperties) { diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 9c021c275..de189f658 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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 inboundBindTarget = new DefaultPollableMessageSource(); + Binding> 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 inboundBindTarget = new DefaultPollableMessageSource(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.setMaxAttempts(2); + properties.setBackOffInitialInterval(0); + properties.getExtension().setAutoBindDlq(true); + Binding> 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 inboundBindTarget = new DefaultPollableMessageSource(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.setMaxAttempts(1); +// properties.getExtension().setRequeueRejected(true); // loops, correctly + properties.getExtension().setAutoBindDlq(true); + Binding> 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 inboundBindTarget = new DefaultPollableMessageSource(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.setMaxAttempts(2); + properties.setBackOffInitialInterval(0); + properties.getExtension().setAutoBindDlq(true); + properties.getExtension().setRepublishToDlq(true); + Binding> 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; diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 430c76232..825ddc84d 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -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, ExtendedProducerProperties> { +public class RabbitTestBinder extends + AbstractPollableConsumerTestBinder, ExtendedProducerProperties> { private final RabbitAdmin rabbitAdmin; @@ -64,7 +68,7 @@ public class RabbitTestBinder extends AbstractTestBinder bindConsumer(String name, String group, MessageChannel moduleInputChannel, ExtendedConsumerProperties properties) { + captureConsumerResources(name, group, properties); + return super.bindConsumer(name, group, moduleInputChannel, properties); + } + + @Override + public Binding> bindPollableConsumer(String name, String group, + PollableSource inboundBindTarget, + ExtendedConsumerProperties properties) { + captureConsumerResources(name, group, properties); + return super.bindPollableConsumer(name, group, inboundBindTarget, properties); + } + + private void captureConsumerResources(String name, String group, + ExtendedConsumerProperties properties) { if (group != null) { if (properties.getExtension().isQueueNameGroupOnly()) { this.queues.add(properties.getExtension().getPrefix() + group); @@ -86,7 +104,6 @@ public class RabbitTestBinder extends AbstractTestBinder