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

SCDF-GH-1528: Fix DLQ Routing Key for SCDF

Browse Source 
Fixes spring-cloud/spring-cloud-dataflow#1528

Data flow creates destinations with name `<stream>.module` and sets the group to the stream name.
The rabbit binder binders a queue `<destination>.<group>` to the `<destination>` 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 `<group>`.

Remove the strip and add back of the prefix to avoid this logic altogether.

Modify one of the tests to simulate SCDF naming.
This commit is contained in:
Gary Russell
2017-07-05 13:44:07 -04:00
committed by Artem Bilan
parent c0a3ab1a7c
commit df33bd0126
2 changed files with 7 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -228,13 +228,7 @@ public class RabbitMessageChannelBinder
@Override
protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> 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 {

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

@@ -846,15 +846,15 @@ public class RabbitBinderTests extends
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqpubtest", "default", moduleInputChannel,
Binding<MessageChannel> 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"));