From 4fda96507ca2960acfd9856ac92302f501c9095d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 31 Aug 2022 13:17:29 -0400 Subject: [PATCH] GH-2453: RMQ: Full Support for Alternate Exchange Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2453 Previously, to configure the use of an alternative exchange (used to route messages when no queue is bound), the user had to manually declare the exchange and any bindings, and modify the destination exchange using a `DeclarablesCustomizer` bean. Add first class support to configure the destination exchange and, optionally, provision the alternate exchange as well as optionally binding a specific queue to it. Resolves #2502 --- binders/rabbit-binder/pom.xml | 2 +- .../properties/RabbitProducerProperties.java | 104 ++++++++++++++++++ .../RabbitExchangeQueueProvisioner.java | 61 ++++++++-- .../RabbitExchangeQueueProvisionerTests.java | 90 +++++++++++++++ .../binder/rabbit/RabbitBinderTests.java | 30 +++++ .../integration/RabbitBinderModuleTests.java | 16 ++- .../main/asciidoc/rabbit/rabbit_overview.adoc | 22 +++- 7 files changed, 310 insertions(+), 15 deletions(-) diff --git a/binders/rabbit-binder/pom.xml b/binders/rabbit-binder/pom.xml index 90f8e3f3f..a0c6d9bfc 100644 --- a/binders/rabbit-binder/pom.xml +++ b/binders/rabbit-binder/pom.xml @@ -53,7 +53,7 @@ org.apache.maven.plugins maven-surefire-plugin - true + false diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index 7b8246e09..46d789657 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java @@ -20,9 +20,11 @@ import java.util.Optional; import jakarta.validation.constraints.Min; +import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -149,6 +151,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties { */ private String streamMessageConverterBeanName; + /** + * Configure an alternate exchange for when no queues are bound. + * @since 4.0 + */ + private AlternateExchange alternateExchange; + /** * @param requestHeaderPatterns the patterns. * @deprecated - use {@link #setHeaderPatterns(String[])}. @@ -302,4 +310,100 @@ public class RabbitProducerProperties extends RabbitCommonProperties { this.streamMessageConverterBeanName = streamMessageConverterBeanName; } + @Nullable + public AlternateExchange getAlternateExchange() { + return this.alternateExchange; + } + + public void setAlternateExchange(AlternateExchange alternate) { + this.alternateExchange = alternate; + } + + public static class AlternateExchange { + + /** + * The alternate exchange name. + */ + private String name; + + /** + * Whether the exchange exists or should be provisioned. + */ + private boolean exists = false; + + /** + * The alternate exchange type. + */ + private String type = ExchangeTypes.TOPIC; + + /** + * Bind a durable queue to the alternate exchange. + */ + private Binding binding; + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + + public boolean isExists() { + return this.exists; + } + + public void setExists(boolean exists) { + this.exists = exists; + } + + public String getType() { + return this.type; + } + + public void setType(String type) { + this.type = type; + } + + public Binding getBinding() { + return this.binding; + } + + public void setBinding(Binding binding) { + this.binding = binding; + } + + public static class Binding { + + /** + * The routing key. + */ + private String routingKey = "#"; + + /** + * The queue name. + */ + private String queue; + + public String getRoutingKey() { + return this.routingKey; + } + + public void setRoutingKey(String routingKey) { + this.routingKey = routingKey; + } + + public String getQueue() { + return this.queue; + } + + public void setQueue(String queue) { + this.queue = queue; + } + + + } + + } + } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 87750dd82..f6f85b44d 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -56,6 +56,7 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonPro import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.AlternateExchange; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.cloud.stream.provisioning.ProvisioningException; @@ -119,9 +120,9 @@ public class RabbitExchangeQueueProvisioner ExtendedProducerProperties producerProperties) { final String exchangeName = applyPrefix( producerProperties.getExtension().getPrefix(), name); - Exchange exchange = buildExchange(producerProperties.getExtension(), - exchangeName); String beanNameQualifier = "prod" + this.producerExchangeBeanNameQualifier.incrementAndGet(); + Exchange exchange = buildExchange(producerProperties.getExtension(), + exchangeName, producerProperties.getExtension().getAlternateExchange(), beanNameQualifier); if (producerProperties.getExtension().isDeclareExchange()) { declareExchange(exchangeName, beanNameQualifier, exchange); } @@ -245,7 +246,7 @@ public class RabbitExchangeQueueProvisioner } String prefix = properties.getExtension().getPrefix(); final String exchangeName = applyPrefix(prefix, name); - Exchange exchange = buildExchange(properties.getExtension(), exchangeName); + Exchange exchange = buildExchange(properties.getExtension(), exchangeName, null, null); if (properties.getExtension().isDeclareExchange()) { declareExchange(exchangeName, anonymous ? anonymousGroup : group, exchange); } @@ -364,25 +365,31 @@ public class RabbitExchangeQueueProvisioner routingKey = "#"; } Map arguments = new HashMap<>(extendedProperties.getQueueBindingArguments()); + return createBinding(exchange, queue, routingKey, arguments, queue.getName()); + } + + private Binding createBinding(Exchange exchange, Queue queue, String routingKey, + @Nullable Map arguments, String beanName) { + if (exchange instanceof TopicExchange) { Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange) .with(routingKey); - declareBinding(queue.getName(), binding); + declareBinding(beanName, binding); return binding; } else if (exchange instanceof DirectExchange) { Binding binding = BindingBuilder.bind(queue).to((DirectExchange) exchange) .with(routingKey); - declareBinding(queue.getName(), binding); + declareBinding(beanName, binding); return binding; } else if (exchange instanceof FanoutExchange) { Binding binding = BindingBuilder.bind(queue).to((FanoutExchange) exchange); - declareBinding(queue.getName(), binding); + declareBinding(beanName, binding); return binding; } else if (exchange instanceof HeadersExchange) { - Binding binding = new Binding(queue.getName(), DestinationType.QUEUE, exchange.getName(), "", arguments); + Binding binding = new Binding(beanName, DestinationType.QUEUE, exchange.getName(), "", arguments); declareBinding(queue.getName(), binding); return binding; } @@ -587,8 +594,9 @@ public class RabbitExchangeQueueProvisioner return prefix + name; } - private Exchange buildExchange(RabbitCommonProperties properties, - String exchangeName) { + private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName, + @Nullable AlternateExchange alternate, @Nullable String beanNameQualifier) { + try { ExchangeBuilder builder = new ExchangeBuilder(exchangeName, properties.getExchangeType()); @@ -599,6 +607,10 @@ public class RabbitExchangeQueueProvisioner if (properties.isDelayedExchange()) { builder.delayed(); } + if (alternate != null && !alternate.isExists()) { + builder.alternate(alternate.getName()); + configureAlternate(alternate, beanNameQualifier); + } return builder.build(); } catch (Exception e) { @@ -606,8 +618,26 @@ public class RabbitExchangeQueueProvisioner } } + private void configureAlternate(AlternateExchange alternate, String beanNameQualifier) { + Exchange exchange = customizeAndDeclare(new ExchangeBuilder(alternate.getName(), alternate.getType()) + .durable(true) + .build()); + addToAutoDeclareContext(alternate.getName() + "." + beanNameQualifier + ".exchange", exchange); + AlternateExchange.Binding binding = alternate.getBinding(); + if (binding != null) { + Queue queue = new Queue(binding.getQueue()); + String beanName = alternate.getName() + "." + binding.getQueue() + "." + beanNameQualifier; + declareQueue(beanName, queue); + Binding toBind = createBinding(exchange, queue, binding.getRoutingKey(), null, beanName); + } + } + private void declareExchange(final String rootName, String group, final Exchange exchangeArg) { - Exchange exchange = exchangeArg; + Exchange exchange = customizeAndDeclare(exchangeArg); + addToAutoDeclareContext(rootName + "." + group + ".exchange", exchange); + } + + private Exchange customizeAndDeclare(Exchange exchange) { for (DeclarableCustomizer customizer : this.customizers) { exchange = (Exchange) customizer.apply(exchange); } @@ -631,7 +661,7 @@ public class RabbitExchangeQueueProvisioner e); } } - addToAutoDeclareContext(rootName + "." + group + ".exchange", exchange); + return exchange; } private void addToAutoDeclareContext(String name, Declarable bean) { @@ -719,6 +749,15 @@ public class RabbitExchangeQueueProvisioner } } } + AlternateExchange alternate = properties.getExtension().getAlternateExchange(); + if (alternate != null) { + removeSingleton(alternate.getName() + "." + qual + ".exchange"); + RabbitProducerProperties.AlternateExchange.Binding binding = alternate.getBinding(); + if (binding != null) { + removeSingleton(alternate.getName() + "." + binding.getQueue() + "." + qual); + removeSingleton(alternate.getName() + "." + binding.getQueue() + "." + qual + ".binding"); + } + } } } } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java index 11ad6ca09..3c00dcadc 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.binder.rabbit.provisioning; import java.io.IOException; +import java.util.Map; import java.util.Set; import com.rabbitmq.client.Channel; @@ -24,6 +25,10 @@ import com.rabbitmq.client.impl.AMQImpl.Queue.DeclareOk; import org.junit.jupiter.api.Test; import org.springframework.amqp.core.Declarables; +import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.core.Exchange; +import org.springframework.amqp.core.ExchangeTypes; +import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.utils.test.TestUtils; @@ -31,6 +36,8 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.AlternateExchange; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.AlternateExchange.Binding; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.context.ApplicationContext; @@ -126,6 +133,89 @@ public class RabbitExchangeQueueProvisionerTests { assertThat(ctx.getBeansOfType(Declarables.class)).isEmpty(); } + @Test + void producerDeclarationsWithAlternateNoQueue() throws IOException { + ConnectionFactory cf = mock(ConnectionFactory.class); + Connection conn = mock(Connection.class); + given(cf.createConnection()).willReturn(conn); + Channel channel = mock(Channel.class); + given(conn.createChannel(anyBoolean())).willReturn(channel); + RabbitExchangeQueueProvisioner provisioner = new RabbitExchangeQueueProvisioner(cf); + + RabbitProducerProperties props = new RabbitProducerProperties(); + AlternateExchange alternate = new AlternateExchange(); + alternate.setName("altEx"); + props.setAlternateExchange(alternate); + ExtendedProducerProperties properties = + new ExtendedProducerProperties(props); + ProducerDestination dest = provisioner.provisionProducerDestination("withAlt", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Map declarables = ctx.getBeansOfType(Declarables.class); + assertThat(declarables).hasSize(2); + String qual = TestUtils.getPropertyValue(dest, "beanNameQualifier", String.class); + Declarables mainEx = declarables.get("withAlt." + qual + ".exchange"); + assertThat(mainEx).isNotNull(); + Exchange exch = (Exchange) mainEx.getDeclarables().iterator().next(); + assertThat(exch.getArguments().get("alternate-exchange")).isEqualTo("altEx"); + Declarables altEx = declarables.get("altEx." + qual + ".exchange"); + assertThat(altEx).isNotNull(); + exch = (Exchange) altEx.getDeclarables().iterator().next(); + assertThat(exch).isInstanceOf(TopicExchange.class); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarables.class)).isEmpty(); + } + + @Test + void producerDeclarationsWithAlternateWithQueue() throws IOException { + ConnectionFactory cf = mock(ConnectionFactory.class); + Connection conn = mock(Connection.class); + given(cf.createConnection()).willReturn(conn); + Channel channel = mock(Channel.class); + willReturn(new DeclareOk("x", 0, 0)) + .given(channel).queueDeclare(any(), eq(Boolean.TRUE), eq(Boolean.FALSE), eq(Boolean.FALSE), any()); + given(conn.createChannel(anyBoolean())).willReturn(channel); + RabbitExchangeQueueProvisioner provisioner = new RabbitExchangeQueueProvisioner(cf); + + RabbitProducerProperties props = new RabbitProducerProperties(); + AlternateExchange alternate = new AlternateExchange(); + alternate.setName("altEx"); + alternate.setType(ExchangeTypes.DIRECT); + Binding binding = new Binding(); + binding.setQueue("altQ"); + binding.setRoutingKey("altRK"); + alternate.setBinding(binding); + props.setAlternateExchange(alternate); + ExtendedProducerProperties properties = + new ExtendedProducerProperties(props); + ProducerDestination dest = provisioner.provisionProducerDestination("withAlt", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Map declarables = ctx.getBeansOfType(Declarables.class); + assertThat(declarables).hasSize(4); + String qual = TestUtils.getPropertyValue(dest, "beanNameQualifier", String.class); + Declarables mainEx = declarables.get("withAlt." + qual + ".exchange"); + assertThat(mainEx).isNotNull(); + Exchange exch = (Exchange) mainEx.getDeclarables().iterator().next(); + assertThat(exch).isInstanceOf(TopicExchange.class); + assertThat(exch.getArguments().get("alternate-exchange")).isEqualTo("altEx"); + Declarables altEx = declarables.get("altEx." + qual + ".exchange"); + assertThat(altEx).isNotNull(); + exch = (Exchange) altEx.getDeclarables().iterator().next(); + assertThat(exch).isInstanceOf(DirectExchange.class); + Declarables queueDec = declarables.get("altEx.altQ." + qual); + assertThat(queueDec).isNotNull(); + Declarables bindingDec = declarables.get("altEx.altQ." + qual + ".binding"); + assertThat(bindingDec).isNotNull(); + org.springframework.amqp.core.Binding bdg = (org.springframework.amqp.core.Binding) bindingDec.getDeclarables() + .iterator().next(); + assertThat(bdg.getExchange()).isEqualTo("altEx"); + assertThat(bdg.getDestination()).isEqualTo("altQ"); + assertThat(bdg.getRoutingKey()).isEqualTo("altRK"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarables.class)).isEmpty(); + } + @Test void producerDeclarationsWithGroupsAndDlq() throws IOException { ConnectionFactory cf = mock(ConnectionFactory.class); diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 1148e743e..8284a8609 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -104,6 +104,7 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonPro import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.AlternateExchange; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.cloud.stream.config.BindingProperties; @@ -388,6 +389,35 @@ public class RabbitBinderTests extends verifyAutoDeclareContextClear(binder); } + @Test + void producerWithAlternateExchange(TestInfo testInfo) throws Exception { + RabbitTestBinder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + ExtendedProducerProperties producerProps = createProducerProperties(testInfo); + AlternateExchange alternate = new AlternateExchange(); + alternate.setName("altEx"); + RabbitProducerProperties.AlternateExchange.Binding binding = + new RabbitProducerProperties.AlternateExchange.Binding(); + binding.setQueue("altQ"); + alternate.setBinding(binding); + producerProps.getExtension().setAlternateExchange(alternate); + Binding producerBinding = binder.bindProducer("alt.0", + moduleOutputChannel, producerProps); + final Message message = MessageBuilder.withPayload("altMessage".getBytes()) + .build(); + moduleOutputChannel.send(message); + producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); + RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource()); + Object received = template.receiveAndConvert("altQ", 10_000); + assertThat(received).isEqualTo("altMessage".getBytes()); + RabbitAdmin admin = new RabbitAdmin(template.getConnectionFactory()); + admin.deleteQueue("altQ"); + admin.deleteExchange("altEx"); + } + @Test public void testConsumerProperties() throws Exception { RabbitTestBinder binder = getBinder(); diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 2e1b36b1b..e6370f9c8 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -62,6 +62,7 @@ import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.AlternateExchange; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer; @@ -100,7 +101,7 @@ public class RabbitBinderModuleTests { private static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(ConnectionFactory.class, Mockito.RETURNS_MOCKS); @RegisterExtension - private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort()); + private final RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort()); private ConfigurableApplicationContext context; @@ -365,7 +366,12 @@ public class RabbitBinderModuleTests { "--spring.cloud.stream.rabbit.default.consumer.exchange-type=direct", "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.batch-size=512", "--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4", - "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.exchange-type=fanout"); + "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.exchange-type=fanout", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.alternateExchange.name=altEx", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.alternate-exchange.exists=true", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.alternate-exchange.type=direct", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.alternate-exchange.binding.queue=altQ", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.alternate-exchange.binding.routing-key=altRK"); BinderFactory binderFactory = context.getBean(BinderFactory.class); Binder rabbitBinder = binderFactory.getBinder(null, MessageChannel.class); @@ -377,6 +383,12 @@ public class RabbitBinderModuleTests { rabbitProducerProperties.getRoutingKeyExpression().getExpressionString()) .isEqualTo("fooRoutingKey"); assertThat(rabbitProducerProperties.getBatchSize()).isEqualTo(512); + AlternateExchange alternate = rabbitProducerProperties.getAlternateExchange(); + assertThat(alternate.getName()).isEqualTo("altEx"); + assertThat(alternate.isExists()).isTrue(); + assertThat(alternate.getType()).isEqualTo("direct"); + assertThat(alternate.getBinding().getQueue()).isEqualTo("altQ"); + assertThat(alternate.getBinding().getRoutingKey()).isEqualTo("altRK"); RabbitConsumerProperties rabbitConsumerProperties = (RabbitConsumerProperties) ((ExtendedPropertiesBinder) rabbitBinder) .getExtendedConsumerProperties("process-in-0"); diff --git a/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc b/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc index 0a2377115..a5a952ba0 100644 --- a/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc +++ b/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc @@ -682,7 +682,27 @@ in the format of `spring.cloud.stream.rabbit.default.=`. Also, keep in mind that binding specific property will override its equivalent in the default. - +altermateExchange.binding.queue:: +If the exchange does not already exist, and a `name` is provided, bind this queue to the alternate exhange. +A simple durable queue with no arguments is provisioned; if more sophisticated configuration is required, you must configure and bind the queue yourself. ++ +Default: `null` +alternateExchange.binding.routingKey +If the exchange does not already exist, and a `name` and `queue` is provided, bind the queue to the alternate exhange using this routing key. ++ +Default: `#` (for the default `topic` alternate exchange) +alternateExchange.exists:: +Whether the alternate exchange exists, or needs to be provisioned. ++ +Default: `false` +alternateExchange.type:: +If the alternate exchange does not already exist, the type of exchange to provision. ++ +Default: `topic` +alternateExchange.name:: +Configure an alternate exchange on the destination exchange. ++ +Default: `null` autoBindDlq:: Whether to automatically declare the DLQ and bind it to the binder DLX. +