customizers) {
+ return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory, customizers);
+ }
+}
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java b/r-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java
new file mode 100644
index 000000000..f31c9a48a
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2021-2021 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.deployer;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+
+/**
+ * Main application class to trigger the deployment of Java Functions based
+ * user application with RabitMQ bindings.
+ *
+ *
+ * It is based on Spring Cloud Function
+ * deploy feature
+ *
+ *
+ * All you need to do is to provide few additional properties to assist Spring Cloud Function deployer.
+ *
+ * For example:
+ *
+ *
+ *
+ * java -jar spring-cloud-stream-binder-rabbit-deployer/target/spring-cloud-stream-binder-rabbit-deployer-3.2.0-SNAPSHOT.jar\
+ * --spring.cloud.function.location=/foo/bar/simplestjar-1.0.0.RELEASE.jar
+ * --spring.cloud.function.function-class=function.example.UpperCaseFunction
+ *
+ *
+ * The only required property is 'spring.cloud.function.location'. Every other property is optional and is based on the style of user application.
+ * For more details on the supported styles of user application please refer to
+ *
+ * Deploying a Packaged Function section of Spring Cloud Function.
+ *
+ * @author Oleg Zhurakousky
+ * @since 3.1.2
+ *
+ */
+@SpringBootApplication
+public class RabbitDeployer {
+
+ public static void main(String[] args) {
+ SpringApplication.run(RabbitDeployer.class, args);
+ }
+
+}
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java b/r-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java
new file mode 100644
index 000000000..91fead450
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021-2021 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.deployer;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.env.EnvironmentPostProcessor;
+import org.springframework.core.env.ConfigurableEnvironment;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ * @since 3.2
+ *
+ */
+class RabbitDeployerEnvironmentPostProcessor implements EnvironmentPostProcessor {
+
+ @Override
+ public void postProcessEnvironment(ConfigurableEnvironment environment,
+ SpringApplication application) {
+ if (!environment.containsProperty("spring.cloud.function.rsocket.enabled")) {
+ environment.getSystemProperties().putIfAbsent("spring.cloud.function.rsocket.enabled", false);
+ }
+ }
+
+}
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders b/r-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders
new file mode 100644
index 000000000..78f94972b
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders
@@ -0,0 +1,2 @@
+rabbit:\
+org.springframework.cloud.stream.binder.rabbit.config.RabbitBinderConfiguration
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories b/r-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories
new file mode 100644
index 000000000..bb946aecc
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,4 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
+org.springframework.cloud.stream.binder.rabbit.config.ExtendedBindingHandlerMappingsProviderConfiguration
+org.springframework.boot.env.EnvironmentPostProcessor:\
+org.springframework.cloud.stream.binder.rabbit.deployer.RabbitDeployerEnvironmentPostProcessor
\ No newline at end of file
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java b/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java
new file mode 100644
index 000000000..acb13785c
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2015-2016 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;
+
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ */
+public class LocalizedQueueConnectionFactoryIntegrationTests {
+
+ @RegisterExtension
+ public static RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
+
+ private LocalizedQueueConnectionFactory lqcf;
+
+ @BeforeEach
+ public void setup() {
+ ConnectionFactory defaultConnectionFactory = rabbitAvailableRule.getResource();
+ String[] addresses = new String[] { "localhost:9999", "localhost:5672" };
+ String[] adminAddresses = new String[] { "http://localhost:15672",
+ "http://localhost:15672" };
+ String[] nodes = new String[] { "foo@bar", "rabbit@localhost" };
+ String vhost = "/";
+ String username = "guest";
+ String password = "guest";
+ this.lqcf = new LocalizedQueueConnectionFactory(defaultConnectionFactory,
+ addresses, adminAddresses, nodes, vhost, username, password, false, null,
+ null, null, null);
+ }
+
+ @Test
+ public void testConnect() {
+ RabbitAdmin admin = new RabbitAdmin(this.lqcf);
+ Queue queue = new Queue(UUID.randomUUID().toString(), false, false, true);
+ admin.declareQueue(queue);
+ ConnectionFactory targetConnectionFactory = this.lqcf
+ .getTargetConnectionFactory("[" + queue.getName() + "]");
+ RabbitTemplate template = new RabbitTemplate(targetConnectionFactory);
+ template.convertAndSend("", queue.getName(), "foo");
+ assertThat(template.receiveAndConvert(queue.getName())).isEqualTo("foo");
+ ((CachingConnectionFactory) targetConnectionFactory).destroy();
+ admin.deleteQueue(queue.getName());
+ }
+
+}
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java b/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java
new file mode 100644
index 000000000..dc5e5cf7e
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java
@@ -0,0 +1,217 @@
+/*
+ * Copyright 2015-2019 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;
+
+import java.net.MalformedURLException;
+import java.net.URISyntaxException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.http.client.Client;
+import com.rabbitmq.http.client.domain.QueueInfo;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import org.springframework.amqp.core.Base64UrlNamingStrategy;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.ChannelCallback;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.cloud.stream.binder.AbstractBinder;
+import org.springframework.cloud.stream.binder.rabbit.admin.RabbitAdminException;
+import org.springframework.cloud.stream.binder.rabbit.admin.RabbitBindingCleaner;
+import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * @author Gary Russell
+ * @since 1.2
+ */
+public class RabbitBinderCleanerTests {
+
+ private static final String BINDER_PREFIX = "binder.";
+
+ private static final Client client;
+
+ static {
+ try {
+ client = new Client("http://localhost:15672/api", "guest", "guest");
+ }
+ catch (MalformedURLException | URISyntaxException e) {
+ throw new RabbitAdminException("Couldn't create a Client", e);
+ }
+ }
+
+ @RegisterExtension
+ public RabbitTestSupport rabbitWithMgmtEnabled = new RabbitTestSupport(true);
+
+ @Test
+ public void testCleanStream() {
+ final RabbitBindingCleaner cleaner = new RabbitBindingCleaner();
+ final String stream1 = new Base64UrlNamingStrategy("foo").generateName();
+ String stream2 = stream1 + "-1";
+ String firstQueue = null;
+ CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource();
+ RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
+ for (int i = 0; i < 5; i++) {
+ String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX,
+ stream1 + ".default." + i);
+ String queue2Name = AbstractBinder.applyPrefix(BINDER_PREFIX,
+ stream2 + ".default." + i);
+ if (firstQueue == null) {
+ firstQueue = queue1Name;
+ }
+ rabbitAdmin.declareQueue(new Queue(queue1Name, true, false, false));
+ rabbitAdmin.declareQueue(new Queue(queue2Name, true, false, false));
+ rabbitAdmin.declareQueue(new Queue(AbstractBinder.constructDLQName(queue1Name), true, false, false));
+ TopicExchange exchange = new TopicExchange(queue1Name);
+ rabbitAdmin.declareExchange(exchange);
+ rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue1Name))
+ .to(exchange).with(queue1Name));
+ exchange = new TopicExchange(queue2Name);
+ rabbitAdmin.declareExchange(exchange);
+ rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue2Name))
+ .to(exchange).with(queue2Name));
+ }
+ final TopicExchange topic1 = new TopicExchange(
+ AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar"));
+ rabbitAdmin.declareExchange(topic1);
+ rabbitAdmin.declareBinding(
+ BindingBuilder.bind(new Queue(firstQueue)).to(topic1).with("#"));
+ String foreignQueue = UUID.randomUUID().toString();
+ rabbitAdmin.declareQueue(new Queue(foreignQueue));
+ rabbitAdmin.declareBinding(
+ BindingBuilder.bind(new Queue(foreignQueue)).to(topic1).with("#"));
+ final TopicExchange topic2 = new TopicExchange(
+ AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar"));
+ rabbitAdmin.declareExchange(topic2);
+ rabbitAdmin.declareBinding(
+ BindingBuilder.bind(new Queue(firstQueue)).to(topic2).with("#"));
+ new RabbitTemplate(connectionFactory).execute(new ChannelCallback() {
+
+ @Override
+ public Void doInRabbit(Channel channel) throws Exception {
+ String queueName = AbstractBinder.applyPrefix(BINDER_PREFIX,
+ stream1 + ".default." + 4);
+ String consumerTag = channel.basicConsume(queueName,
+ new DefaultConsumer(channel));
+ try {
+ waitForConsumerStateNot(queueName, 0);
+ cleaner.clean(stream1, false);
+ fail("Expected exception");
+ }
+ catch (RabbitAdminException e) {
+ assertThat(e)
+ .hasMessageContaining("Queue " + queueName + " is in use");
+ }
+ channel.basicCancel(consumerTag);
+ waitForConsumerStateNot(queueName, 1);
+ try {
+ cleaner.clean(stream1, false);
+ fail("Expected exception");
+ }
+ catch (RabbitAdminException e) {
+ assertThat(e).hasMessageContaining("Cannot delete exchange ");
+ assertThat(e).hasMessageContaining("; it has bindings:");
+ }
+ return null;
+ }
+
+ private void waitForConsumerStateNot(String queueName, long state) throws InterruptedException {
+ int n = 0;
+ QueueInfo queue = client.getQueue("/", queueName);
+ while (n++ < 100 && (queue == null || queue.getConsumerCount() == state)) {
+ Thread.sleep(100);
+ queue = client.getQueue("/", queueName);
+ }
+ assertThat(n).withFailMessage(
+ "Consumer state remained at " + state + " after 10 seconds")
+ .isLessThan(100);
+ }
+
+ });
+ rabbitAdmin.deleteExchange(topic1.getName()); // easier than deleting the binding
+ rabbitAdmin.declareExchange(topic1);
+ rabbitAdmin.deleteQueue(foreignQueue);
+ connectionFactory.destroy();
+ Map> cleanedMap = cleaner.clean(stream1, false);
+ assertThat(cleanedMap).hasSize(2);
+ List cleanedQueues = cleanedMap.get("queues");
+ // should *not* clean stream2
+ assertThat(cleanedQueues).hasSize(10);
+ for (int i = 0; i < 5; i++) {
+ assertThat(cleanedQueues.get(i * 2))
+ .isEqualTo(BINDER_PREFIX + stream1 + ".default." + i);
+ assertThat(cleanedQueues.get(i * 2 + 1))
+ .isEqualTo(BINDER_PREFIX + stream1 + ".default." + i + ".dlq");
+ }
+ List cleanedExchanges = cleanedMap.get("exchanges");
+ assertThat(cleanedExchanges).hasSize(6);
+
+ // wild card *should* clean stream2
+ cleanedMap = cleaner.clean(stream1 + "*", false);
+ assertThat(cleanedMap).hasSize(2);
+ cleanedQueues = cleanedMap.get("queues");
+ assertThat(cleanedQueues).hasSize(5);
+ for (int i = 0; i < 5; i++) {
+ assertThat(cleanedQueues.get(i))
+ .isEqualTo(BINDER_PREFIX + stream2 + ".default." + i);
+ }
+ cleanedExchanges = cleanedMap.get("exchanges");
+ assertThat(cleanedExchanges).hasSize(6);
+ }
+
+ public static class AmqpQueue {
+
+ private boolean autoDelete;
+
+ private boolean durable;
+
+ public AmqpQueue(boolean autoDelete, boolean durable) {
+ this.autoDelete = autoDelete;
+ this.durable = durable;
+ }
+
+ @JsonProperty("auto_delete")
+ protected boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ protected void setAutoDelete(boolean autoDelete) {
+ this.autoDelete = autoDelete;
+ }
+
+ protected boolean isDurable() {
+ return durable;
+ }
+
+ protected void setDurable(boolean durable) {
+ this.durable = durable;
+ }
+
+ }
+
+}
diff --git a/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
new file mode 100644
index 000000000..fe2724561
--- /dev/null
+++ b/r-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
@@ -0,0 +1,2655 @@
+/*
+ * 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.
+ * 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;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.lang.reflect.Constructor;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.zip.Deflater;
+
+import com.rabbitmq.client.LongString;
+import com.rabbitmq.http.client.Client;
+import com.rabbitmq.http.client.domain.BindingInfo;
+import com.rabbitmq.http.client.domain.ExchangeInfo;
+import com.rabbitmq.http.client.domain.QueueInfo;
+import org.apache.commons.logging.Log;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.amqp.AmqpIOException;
+import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
+import org.springframework.amqp.core.AcknowledgeMode;
+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;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.batch.BatchingStrategy;
+import org.springframework.amqp.rabbit.batch.MessageBatch;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
+import org.springframework.amqp.rabbit.connection.RabbitUtils;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.AsyncConsumerStartedEvent;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
+import org.springframework.amqp.utils.test.TestUtils;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.cloud.stream.binder.BinderException;
+import org.springframework.cloud.stream.binder.BinderHeaders;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
+import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
+import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
+import org.springframework.cloud.stream.binder.PartitionTestSupport;
+import org.springframework.cloud.stream.binder.PollableSource;
+import org.springframework.cloud.stream.binder.RequeueCurrentMessageException;
+import org.springframework.cloud.stream.binder.Spy;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties.QuorumConfig;
+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.provisioning.RabbitExchangeQueueProvisioner;
+import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.Lifecycle;
+import org.springframework.expression.spel.standard.SpelExpression;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
+import org.springframework.integration.amqp.support.NackedAmqpMessageException;
+import org.springframework.integration.amqp.support.ReturnedAmqpMessageException;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.handler.BridgeHandler;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.util.ReflectionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Mark Fisher
+ * @author Gary Russell
+ * @author David Turanski
+ * @author Artem Bilan
+ */
+// @checkstyle:off
+public class RabbitBinderTests extends
+ PartitionCapableBinderTests, ExtendedProducerProperties> {
+
+ private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class
+ .getSimpleName();
+
+ public static final String TEST_PREFIX = "bindertest.";
+
+ private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000])
+ .replaceAll("\u0000", "x");
+
+ private int maxStackTraceSize;
+
+
+ @RegisterExtension
+ RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
+
+ @Override
+ protected RabbitTestBinder getBinder() {
+ if (this.testBinder == null) {
+ RabbitProperties rabbitProperties = new RabbitProperties();
+ this.testBinder = new RabbitTestBinder(this.rabbitAvailableRule.getResource(), rabbitProperties);
+ }
+ return this.testBinder;
+ }
+
+ @Override
+ protected ExtendedConsumerProperties createConsumerProperties() {
+ return new ExtendedConsumerProperties<>(new RabbitConsumerProperties());
+ }
+
+ @Override
+ protected ExtendedProducerProperties createProducerProperties(TestInfo testInfo) {
+ ExtendedProducerProperties props = new ExtendedProducerProperties<>(
+ new RabbitProducerProperties());
+
+ if (testInfo.getTestMethod().get().getName().equals("testPartitionedModuleSpEL")) {
+ props.getExtension().setRoutingKeyExpression(
+ spelExpressionParser.parseExpression("'part.0'"));
+ }
+ return props;
+ }
+
+ @Override
+ protected boolean usesExplicitRouting() {
+ return true;
+ }
+
+ @Test
+ public void testSendAndReceiveBad(TestInfo testInfo) throws Exception {
+ RabbitTestBinder binder = getBinder();
+ final AtomicReference event = new AtomicReference<>();
+ binder.getApplicationContext().addApplicationListener(
+ (ApplicationListener) e -> event.set(e));
+ DirectChannel moduleOutputChannel = createBindableChannel("output",
+ new BindingProperties());
+ DirectChannel moduleInputChannel = createBindableChannel("input",
+ new BindingProperties());
+ Binding producerBinding = binder.bindProducer("bad.0",
+ moduleOutputChannel, createProducerProperties(testInfo));
+ assertThat(TestUtils.getPropertyValue(producerBinding,
+ "lifecycle.headersMappedLast", Boolean.class)).isTrue();
+ assertThat(
+ TestUtils
+ .getPropertyValue(producerBinding,
+ "lifecycle.amqpTemplate.messageConverter")
+ .getClass().getName()).contains("Passthrough");
+ ExtendedConsumerProperties consumerProps = createConsumerProperties();
+ consumerProps.getExtension().setContainerType(ContainerType.DIRECT);
+ Binding consumerBinding = binder.bindConsumer("bad.0", "test",
+ moduleInputChannel, consumerProps);
+ assertThat(
+ TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageConverter")
+ .getClass().getName()).contains("Passthrough");
+ assertThat(TestUtils.getPropertyValue(consumerBinding,
+ "lifecycle.messageListenerContainer"))
+ .isInstanceOf(DirectMessageListenerContainer.class);
+ Message message = MessageBuilder.withPayload("bad".getBytes())
+ .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
+ final CountDownLatch latch = new CountDownLatch(3);
+ moduleInputChannel.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ latch.countDown();
+ throw new RuntimeException("bad");
+ }
+ });
+ moduleOutputChannel.send(message);
+ assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(event.get()).isNotNull();
+ producerBinding.unbind();
+ consumerBinding.unbind();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+ @Test
+ public void testProducerErrorChannel(TestInfo testInfo) throws Exception {
+ RabbitTestBinder binder = getBinder();
+ CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+ ccf.setPublisherReturns(true);
+ ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
+ ccf.resetConnection();
+ DirectChannel moduleOutputChannel = createBindableChannel("output",
+ new BindingProperties());
+ ExtendedProducerProperties producerProps = createProducerProperties(testInfo);
+ producerProps.setErrorChannelEnabled(true);
+ Binding producerBinding = binder.bindProducer("ec.0",
+ moduleOutputChannel, producerProps);
+ final Message message = MessageBuilder.withPayload("bad".getBytes())
+ .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
+ SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors",
+ SubscribableChannel.class);
+ final AtomicReference> errorMessage = new AtomicReference<>();
+ final CountDownLatch latch = new CountDownLatch(2);
+ ec.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ errorMessage.set(message);
+ latch.countDown();
+ }
+
+ });
+ SubscribableChannel globalEc = binder.getApplicationContext().getBean(
+ IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
+ SubscribableChannel.class);
+ globalEc.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ latch.countDown();
+ }
+
+ });
+ moduleOutputChannel.send(message);
+ assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class);
+ assertThat(errorMessage.get().getPayload())
+ .isInstanceOf(ReturnedAmqpMessageException.class);
+ ReturnedAmqpMessageException exception = (ReturnedAmqpMessageException) errorMessage
+ .get().getPayload();
+ assertThat(exception.getReplyCode()).isEqualTo(312);
+ assertThat(exception.getReplyText()).isEqualTo("NO_ROUTE");
+
+ AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(producerBinding,
+ "lifecycle", AmqpOutboundEndpoint.class);
+ assertThat(TestUtils.getPropertyValue(endpoint,
+ "confirmCorrelationExpression.expression")).isEqualTo("#root");
+ class WrapperAccessor extends AmqpOutboundEndpoint {
+
+ WrapperAccessor(AmqpTemplate amqpTemplate) {
+ super(amqpTemplate);
+ }
+
+ CorrelationDataWrapper getWrapper() throws Exception {
+ Constructor constructor = CorrelationDataWrapper.class
+ .getDeclaredConstructor(String.class, Object.class,
+ Message.class);
+ ReflectionUtils.makeAccessible(constructor);
+ return constructor.newInstance(UUID.randomUUID().toString(), message, message);
+ }
+
+ }
+ endpoint.confirm(new WrapperAccessor(mock(AmqpTemplate.class)).getWrapper(),
+ false, "Mock NACK");
+ assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class);
+ assertThat(errorMessage.get().getPayload())
+ .isInstanceOf(NackedAmqpMessageException.class);
+ NackedAmqpMessageException nack = (NackedAmqpMessageException) errorMessage.get()
+ .getPayload();
+ assertThat(nack.getNackReason()).isEqualTo("Mock NACK");
+ assertThat(nack.getCorrelationData()).isEqualTo(message);
+ assertThat(nack.getFailedMessage()).isEqualTo(message);
+ producerBinding.unbind();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+ @Test
+ public void testProducerAckChannel(TestInfo testInfo) throws Exception {
+ RabbitTestBinder binder = getBinder();
+ CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+ ccf.setPublisherReturns(true);
+ ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
+ ccf.resetConnection();
+ DirectChannel moduleOutputChannel = createBindableChannel("output",
+ new BindingProperties());
+ ExtendedProducerProperties producerProps = createProducerProperties(testInfo);
+ producerProps.setErrorChannelEnabled(true);
+ producerProps.getExtension().setConfirmAckChannel("acksChannel");
+ Binding producerBinding = binder.bindProducer("acks.0",
+ moduleOutputChannel, producerProps);
+ final Message message = MessageBuilder.withPayload("acksMessage".getBytes())
+ .build();
+ final AtomicReference> confirm = new AtomicReference<>();
+ final CountDownLatch confirmLatch = new CountDownLatch(1);
+ binder.getApplicationContext().getBean("acksChannel", DirectChannel.class)
+ .subscribe(m -> {
+ confirm.set(m);
+ confirmLatch.countDown();
+ });
+ moduleOutputChannel.send(message);
+ assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(confirm.get().getPayload()).isEqualTo("acksMessage".getBytes());
+ producerBinding.unbind();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+ @Test
+ public void testProducerConfirmHeader(TestInfo testInfo) throws Exception {
+ RabbitTestBinder binder = getBinder();
+ CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+ ccf.setPublisherReturns(true);
+ ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
+ ccf.resetConnection();
+ DirectChannel moduleOutputChannel = createBindableChannel("output",
+ new BindingProperties());
+ ExtendedProducerProperties producerProps = createProducerProperties(testInfo);
+ producerProps.getExtension().setUseConfirmHeader(true);
+ Binding producerBinding = binder.bindProducer("confirms.0",
+ moduleOutputChannel, producerProps);
+ CorrelationData correlation = new CorrelationData("testConfirm");
+ final Message message = MessageBuilder.withPayload("confirmsMessage".getBytes())
+ .setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, correlation)
+ .build();
+ moduleOutputChannel.send(message);
+ Confirm confirm = correlation.getFuture().get(10, TimeUnit.SECONDS);
+ assertThat(confirm.isAck()).isTrue();
+ assertThat(correlation.getReturnedMessage()).isNotNull();
+ producerBinding.unbind();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+ @Test
+ public void testConsumerProperties() throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ properties.getExtension().setRequeueRejected(true);
+ properties.getExtension().setTransacted(true);
+ properties.getExtension().setExclusive(true);
+ properties.getExtension().setMissingQueuesFatal(true);
+ properties.getExtension().setFailedDeclarationRetryInterval(1500L);
+ properties.getExtension().setQueueDeclarationRetries(23);
+ Binding consumerBinding = binder.bindConsumer("props.0", null,
+ createBindableChannel("input", new BindingProperties()), properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.AUTO);
+ assertThat(container.getQueueNames()[0])
+ .startsWith(properties.getExtension().getPrefix());
+ assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class))
+ .isTrue();
+ assertThat(TestUtils.getPropertyValue(container, "exclusive", Boolean.class))
+ .isTrue();
+ assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers"))
+ .isEqualTo(1);
+ assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers"))
+ .isNull();
+ assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected",
+ Boolean.class)).isTrue();
+ assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(1);
+ assertThat(TestUtils.getPropertyValue(container, "batchSize")).isEqualTo(1);
+ assertThat(TestUtils.getPropertyValue(container, "missingQueuesFatal",
+ Boolean.class)).isTrue();
+ assertThat(
+ TestUtils.getPropertyValue(container, "failedDeclarationRetryInterval"))
+ .isEqualTo(1500L);
+ assertThat(TestUtils.getPropertyValue(container, "declarationRetries"))
+ .isEqualTo(23);
+ RetryTemplate retry = TestUtils.getPropertyValue(endpoint, "retryTemplate",
+ RetryTemplate.class);
+ assertThat(TestUtils.getPropertyValue(retry, "retryPolicy.maxAttempts"))
+ .isEqualTo(3);
+ assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.initialInterval"))
+ .isEqualTo(1000L);
+ assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.maxInterval"))
+ .isEqualTo(10000L);
+ assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.multiplier"))
+ .isEqualTo(2.0);
+ consumerBinding.unbind();
+ assertThat(endpoint.isRunning()).isFalse();
+
+ properties = createConsumerProperties();
+ properties.getExtension().setAcknowledgeMode(AcknowledgeMode.NONE);
+ properties.setBackOffInitialInterval(2000);
+ properties.setBackOffMaxInterval(20000);
+ properties.setBackOffMultiplier(5.0);
+ properties.setConcurrency(2);
+ properties.setMaxAttempts(23);
+ properties.getExtension().setMaxConcurrency(3);
+ properties.getExtension().setPrefix("foo.");
+ properties.getExtension().setPrefetch(20);
+ properties.getExtension().setHeaderPatterns(new String[] { "foo" });
+ properties.getExtension().setTxSize(10);
+ QuorumConfig quorum = properties.getExtension().getQuorum();
+ quorum.setEnabled(true);
+ quorum.setDeliveryLimit(10);
+ quorum.setInitialGroupSize(1);
+ properties.setInstanceIndex(0);
+ consumerBinding = binder.bindConsumer("props.0", "test",
+ createBindableChannel("input", new BindingProperties()), properties);
+
+ endpoint = extractEndpoint(consumerBinding);
+ container = verifyContainer(endpoint);
+
+ assertThat(container.getQueueNames()[0]).isEqualTo("foo.props.0.test");
+
+ consumerBinding.unbind();
+ assertThat(endpoint.isRunning()).isFalse();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+ @Test
+ public void testMultiplexOnPartitionedConsumer() throws Exception {
+ final ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
+ CachingConnectionFactory cf = new CachingConnectionFactory("localhost",
+ proxy.getPort());
+
+ final RabbitExchangeQueueProvisioner rabbitExchangeQueueProvisioner = new RabbitExchangeQueueProvisioner(cf);
+
+ consumerProperties.setMultiplex(true);
+ consumerProperties.setPartitioned(true);
+ consumerProperties.setInstanceIndexList(Arrays.asList(1, 2, 3));
+
+ final ConsumerDestination consumerDestination = rabbitExchangeQueueProvisioner.provisionConsumerDestination("foo", "boo", consumerProperties);
+
+ final String name = consumerDestination.getName();
+
+ assertThat(name).isEqualTo("foo.boo-1,foo.boo-2,foo.boo-3");
+ }
+
+ @Test
+ public void testMultiplexOnPartitionedConsumerWithMultipleDestinations() throws Exception {
+ final ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
+ CachingConnectionFactory cf = new CachingConnectionFactory("localhost",
+ proxy.getPort());
+
+ final RabbitExchangeQueueProvisioner rabbitExchangeQueueProvisioner = new RabbitExchangeQueueProvisioner(cf);
+
+ consumerProperties.setMultiplex(true);
+ consumerProperties.setPartitioned(true);
+ consumerProperties.setInstanceIndexList(Arrays.asList(1, 2, 3));
+
+ final ConsumerDestination consumerDestination = rabbitExchangeQueueProvisioner.provisionConsumerDestination("foo,qaa", "boo", consumerProperties);
+
+ final String name = consumerDestination.getName();
+
+ assertThat(name).isEqualTo("foo.boo-1,foo.boo-2,foo.boo-3,qaa.boo-1,qaa.boo-2,qaa.boo-3");
+ }
+
+ @Test
+ public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception {
+ RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+ Queue queue = new Queue("propsUser1.infra");
+ admin.declareQueue(queue);
+ DirectExchange exchange = new DirectExchange("propsUser1");
+ admin.declareExchange(exchange);
+ admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo"));
+
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ properties.getExtension().setDeclareExchange(false);
+ properties.getExtension().setBindQueue(false);
+
+ Binding consumerBinding = binder.bindConsumer("propsUser1",
+ "infra", createBindableChannel("input", new BindingProperties()),
+ properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ assertThat(TestUtils.getPropertyValue(container, "missingQueuesFatal",
+ Boolean.class)).isFalse();
+ assertThat(container.isRunning()).isTrue();
+ consumerBinding.unbind();
+ assertThat(container.isRunning()).isFalse();
+ Client client = new Client("http://guest:guest@localhost:15672/api/");
+ List bindings = client.getBindingsBySource("/", exchange.getName());
+ assertThat(bindings.size()).isEqualTo(1);
+ }
+
+ @Test
+ public void testAnonWithBuiltInExchange() throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ properties.getExtension().setDeclareExchange(false);
+ properties.getExtension().setQueueNameGroupOnly(true);
+
+ Binding consumerBinding = binder.bindConsumer("amq.topic", null,
+ createBindableChannel("input", new BindingProperties()), properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ String queueName = container.getQueueNames()[0];
+ assertThat(queueName).startsWith("anonymous.");
+ assertThat(container.isRunning()).isTrue();
+ consumerBinding.unbind();
+ assertThat(container.isRunning()).isFalse();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+
+ @Test
+ public void testAnonWithBuiltInExchangeCustomPrefix() throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ properties.getExtension().setDeclareExchange(false);
+ properties.getExtension().setQueueNameGroupOnly(true);
+ properties.getExtension().setAnonymousGroupPrefix("customPrefix.");
+
+ Binding consumerBinding = binder.bindConsumer("amq.topic", null,
+ createBindableChannel("input", new BindingProperties()), properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ String queueName = container.getQueueNames()[0];
+ assertThat(queueName).startsWith("customPrefix.");
+ assertThat(container.isRunning()).isTrue();
+ consumerBinding.unbind();
+ assertThat(container.isRunning()).isFalse();
+
+ verifyAutoDeclareContextClear(binder);
+ }
+
+
+ @Test
+ public void testConsumerPropertiesWithUserInfrastructureCustomExchangeAndRK()
+ throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ properties.getExtension().setExchangeType(ExchangeTypes.DIRECT);
+ properties.getExtension().setBindingRoutingKey("foo,bar");
+ properties.getExtension().setBindingRoutingKeyDelimiter(",");
+ properties.getExtension().setQueueNameGroupOnly(true);
+ // properties.getExtension().setDelayedExchange(true); // requires delayed message
+ // exchange plugin; tested locally
+
+ String group = "infra";
+ Binding