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

Cherry-pick 2309

Browse Source
This commit is contained in:
Oleg Zhurakousky
2022-03-25 11:21:13 +01:00
parent ac13dd7439
commit 4e66d2b513
6 changed files with 534 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-stream-binder-rabbit-core/pom.xml
View File

@@ -44,5 +44,10 @@
<artifactId>http-client</artifactId>
<version>2.1.0.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-test-support</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

14
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java
View File

@@ -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;
}

174
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -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<DeclarableCustomizer> 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<RabbitConsumerProperties> 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<RabbitConsumerProperties> 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<RabbitProducerProperties> 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 + '}';
}
}
}

221
spring-cloud-stream-binder-rabbit-core/src/test/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisionerTests.java Normal file
View File

@@ -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<RabbitConsumerProperties> properties =
new ExtendedConsumerProperties<RabbitConsumerProperties>(props);
ConsumerDestination dest = provisioner.provisionConsumerDestination("foo", "group", properties);
ApplicationContext ctx =
TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class);
Set<String> 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<RabbitConsumerProperties> properties =
new ExtendedConsumerProperties<RabbitConsumerProperties>(props);
ConsumerDestination dest = provisioner.provisionConsumerDestination("fiz", "group", properties);
ApplicationContext ctx =
TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class);
Set<String> 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<RabbitProducerProperties> properties =
new ExtendedProducerProperties<RabbitProducerProperties>(props);
ProducerDestination dest = provisioner.provisionProducerDestination("bar", properties);
ApplicationContext ctx =
TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class);
Set<String> 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<RabbitProducerProperties> properties =
new ExtendedProducerProperties<RabbitProducerProperties>(props);
properties.setRequiredGroups("group1", "group2");
ProducerDestination dest = provisioner.provisionProducerDestination("baz", properties);
ApplicationContext ctx =
TestUtils.getPropertyValue(provisioner, "autoDeclareContext", ApplicationContext.class);
Set<String> 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<RabbitProducerProperties> properties =
new ExtendedProducerProperties<RabbitProducerProperties>(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<String> 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<RabbitProducerProperties> properties =
new ExtendedProducerProperties<RabbitProducerProperties>(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<String> 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();
}
}

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

@@ -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<RabbitConsumerProperties> consumerProperties) {
provisioningProvider.cleanAutoDeclareContext(consumerDestination,
this.provisioningProvider.cleanAutoDeclareContext(consumerDestination,
consumerProperties);
}
@Override
protected void afterUnbindProducer(ProducerDestination destination,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
this.provisioningProvider.cleanAutoDeclareContext(destination, producerProperties);
}
private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) {
RabbitTemplate rabbitTemplate;
if (properties.isBatchingEnabled()) {

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

@@ -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<RabbitProducerProperties> producerProperties = createProducerProperties(testInfo);
producerProperties.getExtension().setRoutingKey(routingKey);
DirectChannel output = createBindableChannel("output",
createProducerBindingProperties(producerProperties));
output.setBeanName("rkeProducer");
Binding<MessageChannel> 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 {