diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 041a005e4..c106793ef 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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. @@ -16,7 +16,9 @@ package org.springframework.cloud.stream.binder.rabbit.provisioning; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import org.apache.commons.logging.Log; @@ -129,6 +131,23 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener properties) { + if (!properties.isMultiplex()) { + return doProvisionConsumerDestination(name, group, properties); + } + else { + String[] destinations = StringUtils.commaDelimitedListToStringArray(name); + List queues = new ArrayList<>(); + for (String destination : destinations) { + ConsumerDestination dest = doProvisionConsumerDestination(destination.trim(), group, properties); + queues.add(dest.getName()); + } + return new RabbitConsumerDestination( + StringUtils.arrayToCommaDelimitedString(queues.toArray(new String[queues.size()])), null); + } + } + + private ConsumerDestination doProvisionConsumerDestination(String name, String group, + ExtendedConsumerProperties properties) { boolean anonymous = !StringUtils.hasText(group); String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) : properties.getExtension().isQueueNameGroupOnly() ? group : groupedName(name, group); @@ -171,7 +190,7 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener consumerProperties) { synchronized (this.autoDeclareContext) { - removeSingleton(name + ".binding"); - removeSingleton(name); - String dlq = name + ".dlq"; - removeSingleton(dlq + ".binding"); - removeSingleton(dlq); + String[] names = new String[] { destination.getName() }; + if (consumerProperties.isMultiplex()) { + names = StringUtils.commaDelimitedListToStringArray(destination.getName()); + } + for (int i = 0; i < names.length; i++) { + names[i] = names[i].trim(); + } + for (String name : names) { + removeSingleton(name + ".binding"); + removeSingleton(name); + String dlq = name + ".dlq"; + removeSingleton(dlq + ".binding"); + removeSingleton(dlq); + } } } @@ -533,10 +562,10 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener consumerProperties) { + Assert.isTrue(!consumerProperties.isMultiplex(), + "The Spring Integration polled MessageSource does not currently support muiltiple queues"); AmqpMessageSource source = new AmqpMessageSource(this.connectionFactory, destination.getName()); source.setRawMessageHeader(true); return new PolledConsumerResources(source, @@ -568,7 +580,7 @@ public class RabbitMessageChannelBinder @Override protected void afterUnbindConsumer(ConsumerDestination consumerDestination, String group, ExtendedConsumerProperties consumerProperties) { - provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName()); + provisioningProvider.cleanAutoDeclareContext(consumerDestination, consumerProperties); } private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) { 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 9d38e2a08..4e6293e90 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 @@ -49,6 +49,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitManagementTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.AsyncConsumerStartedEvent; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.AmqpHeaders; @@ -641,8 +642,12 @@ public class RabbitBinderTests extends } }); - Binding consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel, - consumerProperties); + consumerProperties.setMultiplex(true); + Binding consumerBinding = binder.bindConsumer("dlqtest,dlqtest2", "default", + moduleInputChannel, consumerProperties); + AbstractMessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding, + "lifecycle.messageListenerContainer", AbstractMessageListenerContainer.class); + assertThat(container.getQueueNames().length).isEqualTo(2); RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo"); @@ -658,6 +663,19 @@ public class RabbitBinderTests extends } assertThat(n).isLessThan(100); + template.convertAndSend("", TEST_PREFIX + "dlqtest2.default", "bar"); + + n = 0; + while (n++ < 100) { + Object deadLetter = template.receiveAndConvert(TEST_PREFIX + "dlqtest2.default.dlq"); + if (deadLetter != null) { + assertThat(deadLetter).isEqualTo("bar"); + break; + } + Thread.sleep(100); + } + assertThat(n).isLessThan(100); + consumerBinding.unbind(); ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.provisioningProvider.autoDeclareContext", @@ -1009,8 +1027,9 @@ public class RabbitBinderTests extends } }); - Binding consumerBinding = binder.bindConsumer("foo.dlqpubtest", "foo", moduleInputChannel, - consumerProperties); + consumerProperties.setMultiplex(true); + Binding consumerBinding = binder.bindConsumer("foo.dlqpubtest,foo.dlqpubtest2", "foo", + moduleInputChannel, consumerProperties); RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo"); @@ -1027,6 +1046,20 @@ public class RabbitBinderTests extends } assertThat(n).isLessThan(100); + template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest2.foo", "bar"); + + n = 0; + while (n++ < 100) { + org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq"); + if (deadLetter != null) { + assertThat(new String(deadLetter.getBody())).isEqualTo("bar"); + assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace")); + break; + } + Thread.sleep(100); + } + assertThat(n).isLessThan(100); + consumerBinding.unbind(); } 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 e8010bf89..4b4a6d382 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 @@ -36,6 +36,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.integration.config.EnableIntegration; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.util.StringUtils; /** * Test support class for {@link RabbitMessageChannelBinder}. @@ -92,15 +93,31 @@ public class RabbitTestBinder extends private void captureConsumerResources(String name, String group, ExtendedConsumerProperties properties) { + String[] names = null; if (group != null) { if (properties.getExtension().isQueueNameGroupOnly()) { this.queues.add(properties.getExtension().getPrefix() + group); } else { - this.queues.add(properties.getExtension().getPrefix() + name + ("." + group)); + if (properties.isMultiplex()) { + names = StringUtils.commaDelimitedListToStringArray(name); + for (String nayme : names) { + this.queues.add(properties.getExtension().getPrefix() + nayme.trim() + "." + group); + } + } + else { + this.queues.add(properties.getExtension().getPrefix() + name + "." + group); + } } } - this.exchanges.add(properties.getExtension().getPrefix() + name); + if (names != null) { + for (String nayme : names) { + this.exchanges.add(properties.getExtension().getPrefix() + nayme.trim()); + } + } + else { + this.exchanges.add(properties.getExtension().getPrefix() + name); + } this.prefixes.add(properties.getExtension().getPrefix()); deadLetters(properties.getExtension()); }