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

GH-2483: RMQ: Fix Redeclare Multi Routing Keys

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2483

https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/242
added support for binding a queue with multiple routing keys. However, only
one binding was added to the `autoRedeclarContext`.

Use the `Declarables` wrapper to support multiple instances with the same
name.

**cherry-pick to 3.2.x**
Resolves #2484
This commit is contained in:
Gary Russell
2022-08-17 12:41:28 -04:00
committed by Oleg Zhurakousky
parent 71ee26b0a1
commit f7146ad02e
2 changed files with 13 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -33,7 +33,9 @@ import org.springframework.amqp.core.Base64UrlNamingStrategy;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.core.DeclarableCustomizer;
import org.springframework.amqp.core.Declarables;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
@@ -632,10 +634,13 @@ public class RabbitExchangeQueueProvisioner
addToAutoDeclareContext(rootName + "." + group + ".exchange", exchange);
}
private void addToAutoDeclareContext(String name, Object bean) {
private void addToAutoDeclareContext(String name, Declarable bean) {
synchronized (this.autoDeclareContext) {
if (!this.autoDeclareContext.containsBean(name)) {
this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean);
this.autoDeclareContext.getBeanFactory().registerSingleton(name, new Declarables(bean));
}
else {
this.autoDeclareContext.getBean(name, Declarables.class).getDeclarables().add(bean);
}
}
}

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

@@ -59,6 +59,7 @@ 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.Declarables;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.MessageDeliveryMode;
@@ -163,7 +164,7 @@ public class RabbitBinderTests extends
private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000]).replaceAll("\u0000", "x");
@RegisterExtension
private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
private final RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
private int maxStackTraceSize;
@@ -602,6 +603,10 @@ public class RabbitBinderTests extends
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
"messageListenerContainer", SimpleMessageListenerContainer.class);
assertThat(container.isRunning()).isTrue();
ApplicationContext ctx =
TestUtils.getPropertyValue(binder, "binder.provisioningProvider.autoDeclareContext",
ApplicationContext.class);
assertThat(ctx.getBean("infra.binding", Declarables.class).getDeclarables()).hasSize(2);
consumerBinding.unbind();
assertThat(container.isRunning()).isFalse();
assertThat(container.getQueueNames()[0]).isEqualTo(group);