From df33bd0126c00f30921baa3752864e1736f70b12 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 5 Jul 2017 13:44:07 -0400 Subject: [PATCH] SCDF-GH-1528: Fix DLQ Routing Key for SCDF Fixes spring-cloud/spring-cloud-dataflow#1528 Data flow creates destinations with name `.module` and sets the group to the stream name. The rabbit binder binders a queue `.` to the `` exchange. If `autoBindDlq` is set, a DLQ is created and bound with key matching the original queue name. There is logic manipulating the destination name to strip off the prefix (if any) because it's re-added later. This logic doesn't work with SCDF because the group occurs twice in the queue name; we end up routing on just the ``. Remove the strip and add back of the prefix to avoid this logic altogether. Modify one of the tests to simulate SCDF naming. --- .../rabbit/RabbitMessageChannelBinder.java | 18 ++++-------------- .../binder/rabbit/RabbitBinderTests.java | 6 +++--- 2 files changed, 7 insertions(+), 17 deletions(-) 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 325172945..29e25613f 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 @@ -228,13 +228,7 @@ public class RabbitMessageChannelBinder @Override protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group, ExtendedConsumerProperties properties) { - - String prefix = properties.getExtension().getPrefix(); String destination = consumerDestination.getName(); - String prefixStripped = (StringUtils.isEmpty(prefix) || !destination.startsWith(prefix)) ? destination - : destination.substring(prefix.length()); - String baseQueueName = StringUtils.hasText(group) ? prefixStripped.substring(0, prefixStripped.indexOf(group)) + group : prefixStripped; - SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer( this.connectionFactory); listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode()); @@ -255,7 +249,7 @@ public class RabbitMessageChannelBinder if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) { RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless() .retryOperations(buildRetryTemplate(properties)) - .recoverer(determineRecoverer(baseQueueName, properties.getExtension())) + .recoverer(determineRecoverer(destination, properties.getExtension())) .build(); listenerContainer.setAdviceChain(retryInterceptor); } @@ -265,7 +259,7 @@ public class RabbitMessageChannelBinder AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); adapter.setBeanFactory(this.getBeanFactory()); - adapter.setBeanName("inbound." + baseQueueName); + adapter.setBeanName("inbound." + destination); DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); mapper.setRequestHeaderNames(properties.getExtension().getHeaderPatterns()); adapter.setHeaderMapper(mapper); @@ -292,9 +286,7 @@ public class RabbitMessageChannelBinder if (properties.isRepublishToDlq()) { RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory); if (properties.getRepublishDeliveyMode() != null) { - return new RepublishMessageRecoverer(errorTemplate, - deadLetterExchangeName(properties), - applyPrefix(properties.getPrefix(), name)) { + return new RepublishMessageRecoverer(errorTemplate, deadLetterExchangeName(properties), name) { @Override public void recover(Message message, Throwable cause) { @@ -305,9 +297,7 @@ public class RabbitMessageChannelBinder }; } else { - return new RepublishMessageRecoverer(errorTemplate, - deadLetterExchangeName(properties), - applyPrefix(properties.getPrefix(), name)); + return new RepublishMessageRecoverer(errorTemplate, deadLetterExchangeName(properties), name); } } else { 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 5e81a33f2..6ef71be3b 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 @@ -846,15 +846,15 @@ public class RabbitBinderTests extends } }); - Binding consumerBinding = binder.bindConsumer("dlqpubtest", "default", moduleInputChannel, + Binding consumerBinding = binder.bindConsumer("foo.dlqpubtest", "foo", moduleInputChannel, consumerProperties); RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); - template.convertAndSend("", TEST_PREFIX + "dlqpubtest.default", "foo"); + template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo"); int n = 0; while (n++ < 100) { - org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "dlqpubtest.default.dlq"); + org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest.foo.dlq"); if (deadLetter != null) { assertThat(new String(deadLetter.getBody())).isEqualTo("foo"); assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));