diff --git a/spring-cloud-stream-binder-rabbit-core/pom.xml b/spring-cloud-stream-binder-rabbit-core/pom.xml index dfa4c8583..7d3e7ee64 100644 --- a/spring-cloud-stream-binder-rabbit-core/pom.xml +++ b/spring-cloud-stream-binder-rabbit-core/pom.xml @@ -44,5 +44,10 @@ http-client 2.1.0.RELEASE + + org.springframework.cloud + spring-cloud-stream-binder-rabbit-test-support + test + diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index bbfd7f64b..b25a6e228 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java @@ -102,6 +102,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties { */ private Expression delayExpression; + /** + * a static routing key when publishing messages; default is the destination name; + * suffixed by "-partition" when partitioned. This is only used if `routingKeyExpression` is null + */ + private String routingKey; + /** * a custom routing key when publishing messages; default is the destination name; * suffixed by "-partition" when partitioned. @@ -239,6 +245,14 @@ public class RabbitProducerProperties extends RabbitCommonProperties { this.routingKeyExpression = routingKeyExpression; } + public String getRoutingKey() { + return this.routingKey; + } + + public void setRoutingKey(String routingKey) { + this.routingKey = routingKey; + } + public String getConfirmAckChannel() { return this.confirmAckChannel; } 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 46264fffc..b0a036b95 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-2018 the original author or authors. + * Copyright 2016-2022 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. @@ -21,6 +21,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Stream; import org.apache.commons.logging.Log; @@ -59,6 +60,7 @@ import org.springframework.cloud.stream.provisioning.ProvisioningException; import org.springframework.cloud.stream.provisioning.ProvisioningProvider; import org.springframework.context.ApplicationListener; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -94,6 +96,8 @@ public class RabbitExchangeQueueProvisioner private final List customizers; + private final AtomicInteger producerExchangeBeanNameQualifier = new AtomicInteger(); + public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory) { this(connectionFactory, Collections.emptyList()); } @@ -115,8 +119,9 @@ public class RabbitExchangeQueueProvisioner producerProperties.getExtension().getPrefix(), name); Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName); + String beanNameQualifier = "prod" + this.producerExchangeBeanNameQualifier.incrementAndGet(); if (producerProperties.getExtension().isDeclareExchange()) { - declareExchange(exchangeName, exchange); + declareExchange(exchangeName, beanNameQualifier, exchange); } Binding binding = null; for (String requiredGroupName : producerProperties.getRequiredGroups()) { @@ -124,7 +129,7 @@ public class RabbitExchangeQueueProvisioner .isQueueNameGroupOnly() ? requiredGroupName : (exchangeName + "." + requiredGroupName); if (!producerProperties.isPartitioned()) { - autoBindDLQ(baseQueueName, baseQueueName, + autoBindDLQ(baseQueueName, baseQueueName, requiredGroupName, producerProperties.getExtension()); if (producerProperties.getExtension().isBindQueue()) { Queue queue = new Queue(baseQueueName, true, false, false, queueArgs( @@ -148,7 +153,7 @@ public class RabbitExchangeQueueProvisioner for (int i = 0; i < producerProperties.getPartitionCount(); i++) { String partitionSuffix = "-" + i; String partitionQueueName = baseQueueName + partitionSuffix; - autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, + autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, requiredGroupName, producerProperties.getExtension()); if (producerProperties.getExtension().isBindQueue()) { Queue queue = new Queue(partitionQueueName, true, false, false, @@ -173,7 +178,7 @@ public class RabbitExchangeQueueProvisioner } } } - return new RabbitProducerDestination(exchange, binding); + return new RabbitProducerDestination(exchange, binding, beanNameQualifier); } @Override @@ -209,7 +214,7 @@ public class RabbitExchangeQueueProvisioner .toArray(String[]::new); consumerDestination = new RabbitConsumerDestination( StringUtils.arrayToCommaDelimitedString(provisionedDestinations), - null); + null, group, name); } return consumerDestination; } @@ -218,21 +223,19 @@ public class RabbitExchangeQueueProvisioner ExtendedConsumerProperties properties) { boolean anonymous = !StringUtils.hasText(group); - Base64UrlNamingStrategy anonQueueNameGenerator = null; + String anonymousGroup = null; if (anonymous) { - anonQueueNameGenerator = new Base64UrlNamingStrategy( + anonymousGroup = new Base64UrlNamingStrategy( properties.getExtension().getAnonymousGroupPrefix() == null ? "" - : properties.getExtension().getAnonymousGroupPrefix()); + : properties.getExtension().getAnonymousGroupPrefix()).generateName(); } String baseQueueName; if (properties.getExtension().isQueueNameGroupOnly()) { - baseQueueName = anonymous ? anonQueueNameGenerator.generateName() - : group; + baseQueueName = anonymous ? anonymousGroup : group; } else { - baseQueueName = groupedName(name, - anonymous ? anonQueueNameGenerator.generateName() : group); + baseQueueName = groupedName(name, anonymous ? anonymousGroup : group); } if (this.logger.isInfoEnabled()) { this.logger.info("declaring queue for inbound: " + baseQueueName @@ -242,7 +245,7 @@ public class RabbitExchangeQueueProvisioner final String exchangeName = applyPrefix(prefix, name); Exchange exchange = buildExchange(properties.getExtension(), exchangeName); if (properties.getExtension().isDeclareExchange()) { - declareExchange(exchangeName, exchange); + declareExchange(exchangeName, anonymous ? anonymousGroup : group, exchange); } String queueName = applyPrefix(prefix, baseQueueName); boolean partitioned = !anonymous && properties.isPartitioned(); @@ -285,9 +288,9 @@ public class RabbitExchangeQueueProvisioner } if (durable) { autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), - queueName, properties.getExtension()); + queueName, group, properties.getExtension()); } - return new RabbitConsumerDestination(queue.getName(), binding); + return new RabbitConsumerDestination(queue.getName(), binding, anonymous ? baseQueueName : group, name); } /** @@ -402,9 +405,10 @@ public class RabbitExchangeQueueProvisioner * @param baseQueueName The base name for the queue (including the binder prefix, if * any). * @param routingKey The routing key for the queue. + * @param group The consumer group. * @param properties the properties. */ - private void autoBindDLQ(final String baseQueueName, String routingKey, + private void autoBindDLQ(final String baseQueueName, String routingKey, String group, RabbitCommonProperties properties) { boolean autoBindDlq = properties.isAutoBindDlq(); if (this.logger.isDebugEnabled()) { @@ -423,7 +427,7 @@ public class RabbitExchangeQueueProvisioner declareQueue(dlqName, dlq); String dlxName = deadLetterExchangeName(properties); if (properties.isDeclareDlx()) { - declareExchange(dlxName, + declareExchange(dlxName, group, new ExchangeBuilder(dlxName, properties.getDeadLetterExchangeType()).durable(true) .build()); @@ -440,7 +444,7 @@ public class RabbitExchangeQueueProvisioner * Also bind with the base queue name when republishToDlq is used, which * does not know about partitioning */ - declareBinding(dlqName, new Binding(dlq.getName(), DestinationType.QUEUE, + declareBinding(dlqName + ".2", new Binding(dlq.getName(), DestinationType.QUEUE, dlxName, baseQueueName, arguments)); } } @@ -600,7 +604,7 @@ public class RabbitExchangeQueueProvisioner } } - private void declareExchange(final String rootName, final Exchange exchangeArg) { + private void declareExchange(final String rootName, String group, final Exchange exchangeArg) { Exchange exchange = exchangeArg; for (DeclarableCustomizer customizer : this.customizers) { exchange = (Exchange) customizer.apply(exchange); @@ -625,7 +629,7 @@ public class RabbitExchangeQueueProvisioner e); } } - addToAutoDeclareContext(rootName + ".exchange", exchange); + addToAutoDeclareContext(rootName + "." + group + ".exchange", exchange); } private void addToAutoDeclareContext(String name, Object bean) { @@ -665,19 +669,93 @@ public class RabbitExchangeQueueProvisioner public void cleanAutoDeclareContext(ConsumerDestination destination, ExtendedConsumerProperties consumerProperties) { + synchronized (this.autoDeclareContext) { Stream.of(StringUtils.tokenizeToStringArray(destination.getName(), ",", true, true)).forEach(name -> { - name = name.trim(); - removeSingleton(name + ".binding"); - removeSingleton(name); - String dlq = name + ".dlq"; - removeSingleton(dlq + ".binding"); - removeSingleton(dlq); + String group = null; + String bindingName = null; + if (destination instanceof RabbitConsumerDestination) { + group = ((RabbitConsumerDestination) destination).getGroup(); + bindingName = ((RabbitConsumerDestination) destination).getBindingName(); + } + RabbitConsumerProperties properties = consumerProperties.getExtension(); + String toRemove = properties.isQueueNameGroupOnly() ? bindingName + "." + group : name.trim(); + boolean partitioned = consumerProperties.isPartitioned(); + if (partitioned) { + toRemove = removePartitionPart(toRemove); + } + removeSingleton(toRemove + ".exchange"); + removeQueueAndBindingBeans(properties, name.trim(), "", group, partitioned); }); } } + public void cleanAutoDeclareContext(ProducerDestination dest, + ExtendedProducerProperties properties) { + + synchronized (this.autoDeclareContext) { + if (dest instanceof RabbitProducerDestination) { + String qual = ((RabbitProducerDestination) dest).getBeanNameQualifier(); + removeSingleton(dest.getName() + "." + qual + ".exchange"); + String[] requiredGroups = properties.getRequiredGroups(); + if (!ObjectUtils.isEmpty(requiredGroups)) { + for (String group : requiredGroups) { + if (properties.isPartitioned()) { + for (int i = 0; i < properties.getPartitionCount(); i++) { + removeQueueAndBindingBeans(properties.getExtension(), + properties.getExtension().isQueueNameGroupOnly() ? "" : dest.getName(), + group + "-" + i, group, true); + } + } + else { + removeQueueAndBindingBeans(properties.getExtension(), dest.getName() + "." + group, "", + group, false); + } + } + } + } + } + } + + private void removeQueueAndBindingBeans(RabbitCommonProperties properties, String name, String suffix, + String group, boolean partitioned) { + + boolean suffixPresent = StringUtils.hasText(suffix); + String withSuffix = name + (suffixPresent ? ("." + suffix) : ""); + String nameDotOptional = name; + if (!StringUtils.hasText(name)) { + withSuffix = suffix; + } + else { + nameDotOptional = name + "."; + } + removeSingleton(withSuffix + ".binding"); + removeSingleton(withSuffix); + String dlq = (suffixPresent ? nameDotOptional + group : withSuffix) + ".dlq"; // only one DLQ when partitioned + if (StringUtils.hasText(properties.getDeadLetterQueueName())) { + dlq = properties.getDeadLetterQueueName(); + } + else if (partitioned) { + String removedPart = removePartitionPart(dlq); + if (!removedPart.endsWith(".dlq")) { + dlq = removedPart + ".dlq"; + } + } + removeSingleton(dlq + ".binding"); + removeSingleton(dlq + ".2.binding"); + removeSingleton(dlq); + removeSingleton(deadLetterExchangeName(properties) + "." + group + ".exchange"); + } + + private String removePartitionPart(String toRemove) { + int finalHyphen = toRemove.lastIndexOf("-"); + if (finalHyphen > 0) { + return toRemove.substring(0, finalHyphen); + } + return toRemove; + } + private void removeSingleton(String name) { if (this.autoDeclareContext.containsBean(name)) { ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext @@ -699,10 +777,13 @@ public class RabbitExchangeQueueProvisioner private final Binding binding; - RabbitProducerDestination(Exchange exchange, Binding binding) { + private final String beanNameQualifier; + + RabbitProducerDestination(Exchange exchange, Binding binding, String beanNameQualifier) { Assert.notNull(exchange, "exchange must not be null"); this.exchange = exchange; this.binding = binding; + this.beanNameQualifier = beanNameQualifier; } @Override @@ -715,10 +796,15 @@ public class RabbitExchangeQueueProvisioner return this.exchange.getName(); } + @Nullable + String getBeanNameQualifier() { + return this.beanNameQualifier; + } + @Override public String toString() { - return "RabbitProducerDestination{" + "exchange=" + exchange + ", binding=" - + binding + '}'; + return "RabbitProducerDestination{" + "exchange=" + this.exchange + ", binding=" + + this.binding + '}'; } } @@ -729,16 +815,16 @@ public class RabbitExchangeQueueProvisioner private final Binding binding; - RabbitConsumerDestination(String queue, Binding binding) { + private final String group; + + private final String bindingName; + + RabbitConsumerDestination(String queue, Binding binding, String group, String bindingName) { Assert.notNull(queue, "queue must not be null"); this.queue = queue; this.binding = binding; - } - - @Override - public String toString() { - return "RabbitConsumerDestination{" + "queue=" + queue + ", binding=" - + binding + '}'; + this.group = group; + this.bindingName = bindingName; } @Override @@ -746,6 +832,20 @@ public class RabbitExchangeQueueProvisioner return this.queue; } + String getGroup() { + return this.group; + } + + String getBindingName() { + return this.bindingName; + } + + @Override + public String toString() { + return "RabbitConsumerDestination{" + "queue=" + this.queue + ", binding=" + + this.binding + ", group=" + this.group + ", bindingName=" + this.bindingName + '}'; + } + } } diff --git a/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java b/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java new file mode 100644 index 000000000..6ac9a26ff --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java @@ -0,0 +1,221 @@ +/* + * Copyright 2022-2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.rabbit.provisioning; + +import java.io.IOException; +import java.util.Set; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.impl.AMQImpl.Queue.DeclareOk; +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.Declarable; +import org.springframework.amqp.rabbit.connection.Connection; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.utils.test.TestUtils; +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.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.context.ApplicationContext; +import org.springframework.expression.common.LiteralExpression; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; + +/** + * @author Gary Russell + * @since 3.2.3 + * + */ +public class RabbitExchangeQueueProvisionerTests { + + @Test + void consumerDeclarationsWithDlq() 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); + + RabbitConsumerProperties props = new RabbitConsumerProperties(); + props.setAutoBindDlq(true); + ExtendedConsumerProperties properties = + new ExtendedConsumerProperties(props); + ConsumerDestination dest = provisioner.provisionConsumerDestination("foo", "group", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Set declarables = ctx.getBeansOfType(Declarable.class).keySet(); + assertThat(declarables).contains("foo.group.exchange", "foo.group", "foo.group.binding", "foo.group.dlq", + "DLX.group.exchange", "foo.group.dlq.binding", "foo.group.dlq.2.binding"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + + @Test + void consumerDeclarationsWithDlqQueueNameIsGroup() 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); + + RabbitConsumerProperties props = new RabbitConsumerProperties(); + props.setAutoBindDlq(true); + props.setQueueNameGroupOnly(true); + ExtendedConsumerProperties properties = + new ExtendedConsumerProperties(props); + ConsumerDestination dest = provisioner.provisionConsumerDestination("fiz", "group", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Set declarables = ctx.getBeansOfType(Declarable.class).keySet(); + assertThat(declarables).contains("fiz.group.exchange", "group", "group.binding", "group.dlq", + "DLX.group.exchange", "group.dlq.binding", "group.dlq.2.binding"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + + @Test + void producerDeclarationsNoGroups() 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(); + ExtendedProducerProperties properties = + new ExtendedProducerProperties(props); + ProducerDestination dest = provisioner.provisionProducerDestination("bar", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Set declarables = ctx.getBeansOfType(Declarable.class).keySet(); + String qual = TestUtils.getPropertyValue(dest, "beanNameQualifier", String.class); + assertThat(declarables).contains("bar." + qual + ".exchange"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + + @Test + void producerDeclarationsWithGroupsAndDlq() 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(); + props.setAutoBindDlq(true); + ExtendedProducerProperties properties = + new ExtendedProducerProperties(props); + properties.setRequiredGroups("group1", "group2"); + ProducerDestination dest = provisioner.provisionProducerDestination("baz", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Set declarables = ctx.getBeansOfType(Declarable.class).keySet(); + String qual = TestUtils.getPropertyValue(dest, "beanNameQualifier", String.class); + assertThat(declarables).contains("baz." + qual + ".exchange", "baz.group1", "baz.group1.binding", + "baz.group1.dlq", "DLX.group1.exchange", "baz.group1.dlq.binding", "baz.group2", "baz.group2.binding", + "baz.group2.dlq", "DLX.group2.exchange", "baz.group2.dlq.binding"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + + @Test + void producerDeclarationsWithGroupsAndDlqAndPartitions() 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(); + props.setAutoBindDlq(true); + ExtendedProducerProperties properties = + new ExtendedProducerProperties(props); + properties.setRequiredGroups("group1", "group2"); + properties.setPartitionKeyExpression(new LiteralExpression("foo")); + properties.setPartitionCount(2); + ProducerDestination dest = provisioner.provisionProducerDestination("qux", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Set declarables = ctx.getBeansOfType(Declarable.class).keySet(); + String qual = TestUtils.getPropertyValue(dest, "beanNameQualifier", String.class); + assertThat(declarables).contains("qux." + qual + ".exchange", "qux.group1-0", "qux.group1-0.binding", + "qux.group1-1", "qux.group1-1.binding", "qux.group1.dlq", "DLX.group1.exchange", + "qux.group1.dlq.binding", "qux.group2-0", + "qux.group2-0.binding", "qux.group2-1", "qux.group2-1.binding", "qux.group2.dlq", "DLX.group2.exchange", + "qux.group2.dlq.binding"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + + @Test + void producerDeclarationsWithGroupsAndDlqAndPartitionsQueueNameIsGroup() 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(); + props.setAutoBindDlq(true); + props.setQueueNameGroupOnly(true); + ExtendedProducerProperties properties = + new ExtendedProducerProperties(props); + properties.setRequiredGroups("group1", "group2"); + properties.setPartitionKeyExpression(new LiteralExpression("foo")); + properties.setPartitionCount(2); + ProducerDestination dest = provisioner.provisionProducerDestination("qux", properties); + ApplicationContext ctx = + TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class); + Set declarables = ctx.getBeansOfType(Declarable.class).keySet(); + String qual = TestUtils.getPropertyValue(dest, "beanNameQualifier", String.class); + assertThat(declarables).contains("qux." + qual + ".exchange", "group1-0", "group1-0.binding", "group1-1", + "group1-1.binding", "group1.dlq", "DLX.group1.exchange", "group1.dlq.binding", "group2-0", + "group2-0.binding", "group2-1", "group2-1.binding", "group2.dlq", "DLX.group2.exchange", + "group2.dlq.binding"); + provisioner.cleanAutoDeclareContext(dest, properties); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + +} 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 6c4ac6e11..6b4fd1633 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-2019 the original author or authors. + * Copyright 2013-2022 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. @@ -217,10 +217,6 @@ public class RabbitMessageChannelBinder extends this.decompressingPostProcessor = decompressingPostProcessor; } - public String getProtocolIdentifier() { - return "amqp"; - } - /** * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress * messages. Defaults to a @@ -964,13 +960,20 @@ public class RabbitMessageChannelBinder extends } @Override - protected void afterUnbindConsumer(ConsumerDestination consumerDestination, - String group, + protected void afterUnbindConsumer(ConsumerDestination consumerDestination, String group, ExtendedConsumerProperties consumerProperties) { - provisioningProvider.cleanAutoDeclareContext(consumerDestination, + + this.provisioningProvider.cleanAutoDeclareContext(consumerDestination, consumerProperties); } + @Override + protected void afterUnbindProducer(ProducerDestination destination, + ExtendedProducerProperties producerProperties) { + + this.provisioningProvider.cleanAutoDeclareContext(destination, producerProperties); + } + private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) { RabbitTemplate rabbitTemplate; if (properties.isBatchingEnabled()) { 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 917e46a23..fe2724561 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-2019 the original author or authors. + * Copyright 2013-2022 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. @@ -57,6 +57,7 @@ import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.AnonymousQueue; import org.springframework.amqp.core.Binding.DestinationType; import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.Declarable; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.core.MessageDeliveryMode; @@ -241,6 +242,8 @@ public class RabbitBinderTests extends assertThat(event.get()).isNotNull(); producerBinding.unbind(); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } @Test @@ -322,6 +325,8 @@ public class RabbitBinderTests extends assertThat(nack.getCorrelationData()).isEqualTo(message); assertThat(nack.getFailedMessage()).isEqualTo(message); producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } @Test @@ -351,6 +356,8 @@ public class RabbitBinderTests extends assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(confirm.get().getPayload()).isEqualTo("acksMessage".getBytes()); producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } @Test @@ -375,6 +382,8 @@ public class RabbitBinderTests extends assertThat(confirm.isAck()).isTrue(); assertThat(correlation.getReturnedMessage()).isNotNull(); producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } @Test @@ -454,6 +463,8 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(endpoint.isRunning()).isFalse(); + + verifyAutoDeclareContextClear(binder); } @Test @@ -543,8 +554,11 @@ public class RabbitBinderTests extends assertThat(container.isRunning()).isTrue(); consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testAnonWithBuiltInExchangeCustomPrefix() throws Exception { RabbitTestBinder binder = getBinder(); @@ -563,8 +577,11 @@ public class RabbitBinderTests extends assertThat(container.isRunning()).isTrue(); consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testConsumerPropertiesWithUserInfrastructureCustomExchangeAndRK() throws Exception { @@ -611,8 +628,11 @@ public class RabbitBinderTests extends assertThat(exchange.getType()).isEqualTo("direct"); assertThat(exchange.isDurable()).isEqualTo(true); assertThat(exchange.isAutoDelete()).isEqualTo(false); + + verifyAutoDeclareContextClear(binder); } + @Test public void testConsumerPropertiesWithUserInfrastructureCustomQueueArgs() throws Exception { @@ -738,8 +758,11 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testConsumerPropertiesWithHeaderExchanges() throws Exception { RabbitTestBinder binder = getBinder(); @@ -788,8 +811,11 @@ public class RabbitBinderTests extends assertThat(bindings.get(0).getDestination()).isEqualTo("propsHeader." + group + ".dlq"); assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any")); assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar")); + + verifyAutoDeclareContextClear(binder); } + @Test public void testProducerProperties(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); @@ -863,8 +889,11 @@ public class RabbitBinderTests extends producerBinding.unbind(); assertThat(endpoint.isRunning()).isFalse(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testDurablePubSubWithAutoBindDLQ() throws Exception { RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); @@ -909,8 +938,11 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq")) .isNotNull(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testNonDurablePubSubWithAutoBindDLQ() throws Exception { RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); @@ -940,8 +972,11 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq")) .isNull(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testAutoBindDLQ() throws Exception { RabbitTestBinder binder = getBinder(); @@ -1012,8 +1047,11 @@ public class RabbitBinderTests extends assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq.binding")) .isFalse(); assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq")).isFalse(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testAutoBindDLQManualAcks() throws Exception { RabbitTestBinder binder = getBinder(); @@ -1087,8 +1125,11 @@ public class RabbitBinderTests extends assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.dlq.binding")) .isFalse(); assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.dlq")).isFalse(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testAutoBindDLQPartionedConsumerFirst(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); @@ -1189,8 +1230,11 @@ public class RabbitBinderTests extends defaultConsumerBinding1.unbind(); defaultConsumerBinding2.unbind(); outputBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test @Disabled public void testAutoBindDLQPartitionedConsumerFirstWithRepublishNoRetry(TestInfo testInfo) @@ -1347,6 +1391,8 @@ public class RabbitBinderTests extends defaultConsumerBinding1.unbind(); defaultConsumerBinding2.unbind(); outputBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } @Test @@ -1451,8 +1497,11 @@ public class RabbitBinderTests extends defaultConsumerBinding1.unbind(); defaultConsumerBinding2.unbind(); outputBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testAutoBindDLQwithRepublish() throws Exception { this.maxStackTraceSize = RabbitUtils @@ -1519,8 +1568,11 @@ public class RabbitBinderTests extends assertThat(template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq")).isNull(); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testAutoBindDLQwithRepublishTx() throws Exception { @@ -1565,8 +1617,11 @@ public class RabbitBinderTests extends assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class)) .isEqualTo(ConfirmType.NONE); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testAutoBindDLQwithRepublishSimpleConfirms() throws Exception { @@ -1613,8 +1668,11 @@ public class RabbitBinderTests extends assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class)) .isEqualTo(ConfirmType.SIMPLE); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testAutoBindDLQwithRepublishCorrelatedConfirms() throws Exception { @@ -1661,8 +1719,11 @@ public class RabbitBinderTests extends assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class)) .isEqualTo(ConfirmType.CORRELATED); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testBatchingAndCompression(TestInfo testInfo) throws Exception { @@ -1723,8 +1784,11 @@ public class RabbitBinderTests extends producerBinding.unbind(); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testProducerBatching(TestInfo testInfo) throws Exception { @@ -1765,8 +1829,11 @@ public class RabbitBinderTests extends producerBinding.unbind(); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testConsumerBatching(TestInfo testInfo) throws Exception { @@ -1803,8 +1870,11 @@ public class RabbitBinderTests extends producerBinding.unbind(); consumerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @SuppressWarnings("unchecked") @Test public void testInternalHeadersNotPropagated(TestInfo testInfo) throws Exception { @@ -1842,8 +1912,11 @@ public class RabbitBinderTests extends producerBinding.unbind(); consumerBinding.unbind(); admin.deleteQueue("propagate"); + + verifyAutoDeclareContextClear(binder); } + /* * Test late binding due to broker down; queues with and without DLQs, and partitioned * queues. @@ -1980,8 +2053,11 @@ public class RabbitBinderTests extends cf.destroy(); this.rabbitAvailableRule.getResource().destroy(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testBadUserDeclarationsFatal() throws Exception { RabbitTestBinder binder = getBinder(); @@ -2019,8 +2095,11 @@ public class RabbitBinderTests extends binding.unbind(); } } + + verifyAutoDeclareContextClear(binder); } + @Test public void testRoutingKeyExpression(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); @@ -2062,8 +2141,46 @@ public class RabbitBinderTests extends .isEqualTo("{\"field\":\"rkeTest\"}"); producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + + @Test + public void testRoutingKey(TestInfo testInfo) throws Exception { + String routingKey = "static.key"; + RabbitTestBinder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(testInfo); + producerProperties.getExtension().setRoutingKey(routingKey); + + DirectChannel output = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + output.setBeanName("rkeProducer"); + Binding producerBinding = binder.bindProducer("rke", output, + producerProperties); + + RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + Queue queue = new AnonymousQueue(); + DirectExchange exchange = new DirectExchange("rke"); + org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue) + .to(exchange).with(routingKey); + admin.declareQueue(queue); + admin.declareBinding(binding); + + output.send(new GenericMessage<>(new Pojo("rkeTest"))); + + Object out = spyOn(queue.getName()).receive(false); + assertThat(out).isInstanceOf(byte[].class); + assertThat(new String((byte[]) out, StandardCharsets.UTF_8)) + .isEqualTo("{\"field\":\"rkeTest\"}"); + + + producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); + } + + @Test public void testRoutingKeyExpressionPartitionedAndDelay(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); @@ -2113,8 +2230,11 @@ public class RabbitBinderTests extends .isEqualTo("{\"field\":\"rkepTest\"}"); producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testPolledConsumer() throws Exception { RabbitTestBinder binder = getBinder(); @@ -2136,8 +2256,11 @@ public class RabbitBinderTests extends } assertThat(polled).isTrue(); binding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testPolledConsumerRequeue() throws Exception { RabbitTestBinder binder = getBinder(); @@ -2167,8 +2290,11 @@ public class RabbitBinderTests extends }); assertThat(polled).isTrue(); binding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testPolledConsumerWithDlq() throws Exception { RabbitTestBinder binder = getBinder(); @@ -2201,8 +2327,11 @@ public class RabbitBinderTests extends .receive("pollableDlq.group.dlq", 10_000); assertThat(deadLetter).isNotNull(); binding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testPolledConsumerWithDlqNoRetry() throws Exception { RabbitTestBinder binder = getBinder(); @@ -2233,8 +2362,11 @@ public class RabbitBinderTests extends .receive("pollableDlqNoRetry.group.dlq", 10_000); assertThat(deadLetter).isNotNull(); binding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testPolledConsumerWithDlqRePub() throws Exception { RabbitTestBinder binder = getBinder(); @@ -2263,8 +2395,11 @@ public class RabbitBinderTests extends .receive("pollableDlqRePub.group.dlq", 10_000); assertThat(deadLetter).isNotNull(); binding.unbind(); + + verifyAutoDeclareContextClear(binder); } + @Test public void testCustomBatchingStrategy(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); @@ -2300,8 +2435,11 @@ public class RabbitBinderTests extends assertThat(new String((byte[]) out)).isEqualTo("0\u0000\n1\u0000\n2\u0000\n3\u0000\n4\u0000\n"); producerBinding.unbind(); + + verifyAutoDeclareContextClear(binder); } + private SimpleMessageListenerContainer verifyContainer(Lifecycle endpoint) { SimpleMessageListenerContainer container; RetryTemplate retry; @@ -2415,6 +2553,13 @@ public class RabbitBinderTests extends return stringWriter.getBuffer().toString(); } + private void verifyAutoDeclareContextClear(RabbitTestBinder binder) { + ApplicationContext ctx = + TestUtils.getPropertyValue(binder, "binder.provisioningProvider.autoDeclareContext", + ApplicationContext.class); + assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty(); + } + public static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy {