customizers) {
+ return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory, customizers);
+ }
+}
diff --git a/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java b/rabbit-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/rabbit-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/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java b/rabbit-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/rabbit-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/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders b/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders
new file mode 100644
index 000000000..78f94972b
--- /dev/null
+++ b/rabbit-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/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories b/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories
new file mode 100644
index 000000000..bb946aecc
--- /dev/null
+++ b/rabbit-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/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java b/rabbit-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/rabbit-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/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java b/rabbit-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/rabbit-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/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
new file mode 100644
index 000000000..de40c4de8
--- /dev/null
+++ b/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
@@ -0,0 +1,2542 @@
+/*
+ * Copyright 2013-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.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.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();
+ }
+
+ @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();
+ }
+
+ @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();
+ }
+
+ @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();
+ }
+
+ @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();
+ }
+
+ @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();
+ }
+
+ @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();
+ }
+
+ @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 consumerBinding = binder.bindConsumer("propsUser2", group,
+ createBindableChannel("input", new BindingProperties()), properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ assertThat(container.isRunning()).isTrue();
+ consumerBinding.unbind();
+ assertThat(container.isRunning()).isFalse();
+ assertThat(container.getQueueNames()[0]).isEqualTo(group);
+ Client client = new Client("http://guest:guest@localhost:15672/api/");
+ List bindings = client.getBindingsBySource("/", "propsUser2");
+ int n = 0;
+ while (n++ < 100 && bindings == null || bindings.size() < 1) {
+ Thread.sleep(100);
+ bindings = client.getBindingsBySource("/", "propsUser2");
+ }
+ assertThat(bindings.size()).isEqualTo(2);
+ assertThat(bindings.get(0).getSource()).isEqualTo("propsUser2");
+ assertThat(bindings.get(0).getDestination()).isEqualTo(group);
+ assertThat(bindings.get(0).getRoutingKey()).isIn("foo", "bar");
+ assertThat(bindings.get(1).getSource()).isEqualTo("propsUser2");
+ assertThat(bindings.get(1).getDestination()).isEqualTo(group);
+ assertThat(bindings.get(1).getRoutingKey()).isIn("foo", "bar");
+ assertThat(bindings.get(1).getRoutingKey()).isNotEqualTo(bindings.get(0).getRoutingKey());
+
+ ExchangeInfo exchange = client.getExchange("/", "propsUser2");
+ while (n++ < 100 && exchange == null) {
+ Thread.sleep(100);
+ exchange = client.getExchange("/", "propsUser2");
+ }
+ assertThat(exchange.getType()).isEqualTo("direct");
+ assertThat(exchange.isDurable()).isEqualTo(true);
+ assertThat(exchange.isAutoDelete()).isEqualTo(false);
+ }
+
+ @Test
+ public void testConsumerPropertiesWithUserInfrastructureCustomQueueArgs()
+ throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ RabbitConsumerProperties extProps = properties.getExtension();
+ extProps.setExchangeType(ExchangeTypes.DIRECT);
+ extProps.setExchangeDurable(false);
+ extProps.setExchangeAutoDelete(true);
+ extProps.setBindingRoutingKey("foo");
+ extProps.setExpires(30_000);
+ extProps.setLazy(true);
+ extProps.setMaxLength(10_000);
+ extProps.setMaxLengthBytes(100_000);
+ extProps.setMaxPriority(10);
+ extProps.setOverflowBehavior("drop-head");
+ extProps.setTtl(2_000);
+ extProps.setAutoBindDlq(true);
+ extProps.setDeadLetterQueueName("customDLQ");
+ extProps.setDeadLetterExchange("customDLX");
+ extProps.setDeadLetterExchangeType(ExchangeTypes.TOPIC);
+ extProps.setDeadLetterRoutingKey("customDLRK");
+ extProps.setDlqDeadLetterExchange("propsUser3");
+ // GH-259 - if the next line was commented, the test failed.
+ extProps.setDlqDeadLetterRoutingKey("propsUser3");
+ extProps.setDlqExpires(60_000);
+ extProps.setDlqLazy(true);
+ extProps.setDlqMaxLength(20_000);
+ extProps.setDlqMaxLengthBytes(40_000);
+ extProps.setDlqOverflowBehavior("reject-publish");
+ extProps.setDlqMaxPriority(8);
+ extProps.setDlqTtl(1_000);
+ extProps.setConsumerTagPrefix("testConsumerTag");
+ extProps.setExclusive(true);
+
+ Binding consumerBinding = binder.bindConsumer("propsUser3",
+ "infra", createBindableChannel("input", new BindingProperties()),
+ properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ assertThat(container.isRunning()).isTrue();
+ Client client = new Client("http://guest:guest@localhost:15672/api");
+ List bindings = client.getBindingsBySource("/", "propsUser3");
+ int n = 0;
+ while (n++ < 100 && bindings == null || bindings.size() < 1) {
+ Thread.sleep(100);
+ bindings = client.getBindingsBySource("/", "propsUser3");
+ }
+ assertThat(bindings.size()).isEqualTo(1);
+ assertThat(bindings.get(0).getSource()).isEqualTo("propsUser3");
+ assertThat(bindings.get(0).getDestination()).isEqualTo("propsUser3.infra");
+ assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo");
+
+ bindings = client.getBindingsBySource("/", "customDLX");
+ n = 0;
+ while (n++ < 100 && bindings == null || bindings.size() < 1) {
+ Thread.sleep(100);
+ bindings = client.getBindingsBySource("/", "customDLX");
+ }
+// assertThat(bindings.size()).isEqualTo(1);
+ assertThat(bindings.get(0).getSource()).isEqualTo("customDLX");
+ assertThat(bindings.get(0).getDestination()).isEqualTo("customDLQ");
+ assertThat(bindings.get(0).getRoutingKey()).isEqualTo("customDLRK");
+
+ ExchangeInfo exchange = client.getExchange("/", "propsUser3");
+ n = 0;
+ while (n++ < 100 && exchange == null) {
+ Thread.sleep(100);
+ exchange = client.getExchange("/", "propsUser3");
+ }
+ assertThat(exchange.getType()).isEqualTo("direct");
+ assertThat(exchange.isDurable()).isEqualTo(false);
+ assertThat(exchange.isAutoDelete()).isEqualTo(true);
+
+ exchange = client.getExchange("/", "customDLX");
+ n = 0;
+ while (n++ < 100 && exchange == null) {
+ Thread.sleep(100);
+ exchange = client.getExchange("/", "customDLX");
+ }
+ assertThat(exchange.getType()).isEqualTo("topic");
+ assertThat(exchange.isDurable()).isEqualTo(true);
+ assertThat(exchange.isAutoDelete()).isEqualTo(false);
+
+ QueueInfo queue = client.getQueue("/", "propsUser3.infra");
+ n = 0;
+ while (n++ < 100 && queue == null || queue.getConsumerCount() == 0) {
+ Thread.sleep(100);
+ queue = client.getQueue("/", "propsUser3.infra");
+ }
+ assertThat(queue).isNotNull();
+ Map args = queue.getArguments();
+ assertThat(args.get("x-expires")).isEqualTo(30_000);
+ assertThat(args.get("x-max-length")).isEqualTo(10_000);
+ assertThat(args.get("x-max-length-bytes")).isEqualTo(100_000);
+ assertThat(args.get("x-overflow")).isEqualTo("drop-head");
+ assertThat(args.get("x-max-priority")).isEqualTo(10);
+ assertThat(args.get("x-message-ttl")).isEqualTo(2_000);
+ assertThat(args.get("x-dead-letter-exchange")).isEqualTo("customDLX");
+ assertThat(args.get("x-dead-letter-routing-key")).isEqualTo("customDLRK");
+ assertThat(args.get("x-queue-mode")).isEqualTo("lazy");
+ assertThat(queue.getExclusiveConsumerTag()).isEqualTo("testConsumerTag#0");
+
+ queue = client.getQueue("/", "customDLQ");
+
+ n = 0;
+ while (n++ < 100 && queue == null) {
+ Thread.sleep(100);
+ queue = client.getQueue("/", "customDLQ");
+ }
+ assertThat(queue).isNotNull();
+ args = queue.getArguments();
+ assertThat(args.get("x-expires")).isEqualTo(60_000);
+ assertThat(args.get("x-max-length")).isEqualTo(20_000);
+ assertThat(args.get("x-max-length-bytes")).isEqualTo(40_000);
+ assertThat(args.get("x-overflow")).isEqualTo("reject-publish");
+ assertThat(args.get("x-max-priority")).isEqualTo(8);
+ assertThat(args.get("x-message-ttl")).isEqualTo(1_000);
+ assertThat(args.get("x-dead-letter-exchange")).isEqualTo("propsUser3");
+ assertThat(args.get("x-dead-letter-routing-key")).isEqualTo("propsUser3");
+ assertThat(args.get("x-queue-mode")).isEqualTo("lazy");
+
+ consumerBinding.unbind();
+ assertThat(container.isRunning()).isFalse();
+ }
+
+ @Test
+ public void testConsumerPropertiesWithHeaderExchanges() throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties properties = createConsumerProperties();
+ properties.getExtension().setExchangeType(ExchangeTypes.HEADERS);
+ properties.getExtension().setAutoBindDlq(true);
+ properties.getExtension().setDeadLetterExchange(ExchangeTypes.HEADERS);
+ properties.getExtension().setDeadLetterExchange("propsHeader.dlx");
+ Map queueBindingArguments = new HashMap<>();
+ queueBindingArguments.put("x-match", "any");
+ queueBindingArguments.put("foo", "bar");
+ properties.getExtension().setQueueBindingArguments(queueBindingArguments);
+ properties.getExtension().setDlqBindingArguments(queueBindingArguments);
+
+ String group = "bindingArgs";
+ Binding consumerBinding = binder.bindConsumer("propsHeader", group,
+ createBindableChannel("input", new BindingProperties()), properties);
+ Lifecycle endpoint = extractEndpoint(consumerBinding);
+ SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
+ "messageListenerContainer", SimpleMessageListenerContainer.class);
+ assertThat(container.isRunning()).isTrue();
+ consumerBinding.unbind();
+ assertThat(container.isRunning()).isFalse();
+ assertThat(container.getQueueNames()[0]).isEqualTo("propsHeader." + group);
+ Client client = new Client("http://guest:guest@localhost:15672/api/");
+ List bindings = client.getBindingsBySource("/", "propsHeader");
+ int n = 0;
+ while (n++ < 100 && bindings == null || bindings.size() < 1) {
+ Thread.sleep(100);
+ bindings = client.getBindingsBySource("/", "propsHeader");
+ }
+ assertThat(bindings.size()).isEqualTo(1);
+ assertThat(bindings.get(0).getSource()).isEqualTo("propsHeader");
+ assertThat(bindings.get(0).getDestination()).isEqualTo("propsHeader." + group);
+ 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"));
+
+ bindings = client.getBindingsBySource("/", "propsHeader.dlx");
+ n = 0;
+ while (n++ < 100 && bindings == null || bindings.size() < 1) {
+ Thread.sleep(100);
+ bindings = client.getBindingsBySource("/", "propsHeader.dlx");
+ }
+ assertThat(bindings.size()).isEqualTo(1);
+ assertThat(bindings.get(0).getSource()).isEqualTo("propsHeader.dlx");
+ 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"));
+ }
+
+ @Test
+ public void testProducerProperties(TestInfo testInfo) throws Exception {
+ RabbitTestBinder binder = getBinder();
+ Binding producerBinding = binder.bindProducer("props.0",
+ createBindableChannel("input", new BindingProperties()),
+ createProducerProperties(testInfo));
+ Lifecycle endpoint = extractEndpoint(producerBinding);
+ MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint,
+ "defaultDeliveryMode", MessageDeliveryMode.class);
+ assertThat(mode).isEqualTo(MessageDeliveryMode.PERSISTENT);
+ List requestHeaders = TestUtils.getPropertyValue(endpoint,
+ "headerMapper.requestHeaderMatcher.matchers", List.class);
+ assertThat(requestHeaders).hasSize(5);
+ producerBinding.unbind();
+ assertThat(endpoint.isRunning()).isFalse();
+ assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional",
+ Boolean.class)).isFalse();
+
+ ExtendedProducerProperties producerProperties = createProducerProperties(testInfo);
+ this.applicationContext.registerBean("pkExtractor",
+ TestPartitionKeyExtractorClass.class, () -> new TestPartitionKeyExtractorClass());
+ this.applicationContext.registerBean("pkSelector",
+ TestPartitionSelectorClass.class, () -> new TestPartitionSelectorClass());
+ producerProperties.setPartitionKeyExtractorName("pkExtractor");
+ producerProperties.setPartitionSelectorName("pkSelector");
+ producerProperties.getExtension().setPrefix("foo.");
+ producerProperties.getExtension()
+ .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+ producerProperties.getExtension().setHeaderPatterns(new String[] { "foo" });
+ producerProperties
+ .setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'"));
+ producerProperties.setPartitionSelectorExpression(
+ spelExpressionParser.parseExpression("0"));
+ producerProperties.setPartitionCount(1);
+ producerProperties.getExtension().setTransacted(true);
+ producerProperties.getExtension()
+ .setDelayExpression(spelExpressionParser.parseExpression("42"));
+ producerProperties.setRequiredGroups("prodPropsRequired");
+
+ BindingProperties producerBindingProperties = createProducerBindingProperties(
+ producerProperties);
+ DirectChannel channel = createBindableChannel("output",
+ producerBindingProperties);
+ producerBinding = binder.bindProducer("props.0", channel, producerProperties);
+
+ ConnectionFactory producerConnectionFactory = TestUtils.getPropertyValue(
+ producerBinding, "lifecycle.amqpTemplate.connectionFactory",
+ ConnectionFactory.class);
+
+ assertThat(this.rabbitAvailableRule.getResource())
+ .isSameAs(producerConnectionFactory);
+
+ endpoint = extractEndpoint(producerBinding);
+ assertThat(getEndpointRouting(endpoint)).isEqualTo(
+ "'props.0-' + headers['" + BinderHeaders.PARTITION_HEADER + "']");
+ assertThat(TestUtils
+ .getPropertyValue(endpoint, "delayExpression", SpelExpression.class)
+ .getExpressionString()).isEqualTo("42");
+ mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode",
+ MessageDeliveryMode.class);
+ assertThat(mode).isEqualTo(MessageDeliveryMode.NON_PERSISTENT);
+ assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional",
+ Boolean.class)).isTrue();
+ verifyFooRequestProducer(endpoint);
+ channel.send(new GenericMessage<>("foo"));
+ org.springframework.amqp.core.Message received = new RabbitTemplate(
+ this.rabbitAvailableRule.getResource())
+ .receive("foo.props.0.prodPropsRequired-0", 10_000);
+ assertThat(received).isNotNull();
+ assertThat(received.getMessageProperties().getReceivedDelay()).isEqualTo(42);
+
+ producerBinding.unbind();
+ assertThat(endpoint.isRunning()).isFalse();
+ }
+
+ @Test
+ public void testDurablePubSubWithAutoBindDLQ() throws Exception {
+ RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+
+ RabbitTestBinder binder = getBinder();
+
+ ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ consumerProperties.getExtension().setPrefix(TEST_PREFIX);
+ consumerProperties.getExtension().setAutoBindDlq(true);
+ consumerProperties.getExtension().setDurableSubscription(true);
+ consumerProperties.setMaxAttempts(1); // disable retry
+ DirectChannel moduleInputChannel = createBindableChannel("input",
+ createConsumerBindingProperties(consumerProperties));
+ moduleInputChannel.setBeanName("durableTest");
+ moduleInputChannel.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ throw new RuntimeException("foo");
+ }
+
+ });
+ Binding consumerBinding = binder.bindConsumer("durabletest.0",
+ "tgroup", moduleInputChannel, consumerProperties);
+
+ RabbitTemplate template = new RabbitTemplate(
+ this.rabbitAvailableRule.getResource());
+ template.convertAndSend(TEST_PREFIX + "durabletest.0", "", "foo");
+
+ int n = 0;
+ while (n++ < 100) {
+ Object deadLetter = template
+ .receiveAndConvert(TEST_PREFIX + "durabletest.0.tgroup.dlq");
+ if (deadLetter != null) {
+ assertThat(deadLetter).isEqualTo("foo");
+ break;
+ }
+ Thread.sleep(100);
+ }
+ assertThat(n).isLessThan(100);
+
+ consumerBinding.unbind();
+ assertThat(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq"))
+ .isNotNull();
+ }
+
+ @Test
+ public void testNonDurablePubSubWithAutoBindDLQ() throws Exception {
+ RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ consumerProperties.getExtension().setPrefix(TEST_PREFIX);
+ consumerProperties.getExtension().setAutoBindDlq(true);
+ consumerProperties.getExtension().setDurableSubscription(false);
+ consumerProperties.setMaxAttempts(1); // disable retry
+ BindingProperties bindingProperties = createConsumerBindingProperties(
+ consumerProperties);
+ DirectChannel moduleInputChannel = createBindableChannel("input",
+ bindingProperties);
+ moduleInputChannel.setBeanName("nondurabletest");
+ moduleInputChannel.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ throw new RuntimeException("foo");
+ }
+
+ });
+ Binding consumerBinding = binder.bindConsumer("nondurabletest.0",
+ "tgroup", moduleInputChannel, consumerProperties);
+
+ consumerBinding.unbind();
+ assertThat(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq"))
+ .isNull();
+ }
+
+ @Test
+ public void testAutoBindDLQ() throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ consumerProperties.getExtension().setPrefix(TEST_PREFIX);
+ consumerProperties.getExtension().setAutoBindDlq(true);
+ consumerProperties.setMaxAttempts(1); // disable retry
+ consumerProperties.getExtension().setDurableSubscription(true);
+ BindingProperties bindingProperties = createConsumerBindingProperties(
+ consumerProperties);
+ DirectChannel moduleInputChannel = createBindableChannel("input",
+ bindingProperties);
+ moduleInputChannel.setBeanName("dlqTest");
+ moduleInputChannel.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ throw new RuntimeException("foo");
+ }
+
+ });
+ consumerProperties.setMultiplex(true);
+ Binding consumerBinding = binder.bindConsumer("dlqtest,dlqtest2",
+ "default", moduleInputChannel, consumerProperties);
+ AbstractMessageListenerContainer container = TestUtils.getPropertyValue(
+ consumerBinding, "lifecycle.messageListenerContainer",
+ AbstractMessageListenerContainer.class);
+ assertThat(container.getQueueNames().length).isEqualTo(2);
+
+ RabbitTemplate template = new RabbitTemplate(
+ this.rabbitAvailableRule.getResource());
+ template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo");
+
+ int n = 0;
+ while (n++ < 100) {
+ Object deadLetter = template
+ .receiveAndConvert(TEST_PREFIX + "dlqtest.default.dlq");
+ if (deadLetter != null) {
+ assertThat(deadLetter).isEqualTo("foo");
+ break;
+ }
+ Thread.sleep(100);
+ }
+ assertThat(n).isLessThan(100);
+
+ template.convertAndSend("", TEST_PREFIX + "dlqtest2.default", "bar");
+
+ n = 0;
+ while (n++ < 100) {
+ Object deadLetter = template
+ .receiveAndConvert(TEST_PREFIX + "dlqtest2.default.dlq");
+ if (deadLetter != null) {
+ assertThat(deadLetter).isEqualTo("bar");
+ break;
+ }
+ Thread.sleep(100);
+ }
+ assertThat(n).isLessThan(100);
+
+ consumerBinding.unbind();
+
+ ApplicationContext context = TestUtils.getPropertyValue(binder,
+ "binder.provisioningProvider.autoDeclareContext",
+ ApplicationContext.class);
+ assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.binding"))
+ .isFalse();
+ assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default")).isFalse();
+ assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq.binding"))
+ .isFalse();
+ assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq")).isFalse();
+ }
+
+ @Test
+ public void testAutoBindDLQManualAcks() throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ consumerProperties.getExtension().setPrefix(TEST_PREFIX);
+ consumerProperties.getExtension().setAutoBindDlq(true);
+ consumerProperties.setMaxAttempts(2);
+ consumerProperties.getExtension().setDurableSubscription(true);
+ consumerProperties.getExtension().setAcknowledgeMode(AcknowledgeMode.MANUAL);
+ BindingProperties bindingProperties = createConsumerBindingProperties(
+ consumerProperties);
+ DirectChannel moduleInputChannel = createBindableChannel("input",
+ bindingProperties);
+ moduleInputChannel.setBeanName("dlqTestManual");
+ Client client = new Client("http://guest:guest@localhost:15672/api");
+ moduleInputChannel.subscribe(new MessageHandler() {
+
+ @Override
+ public void handleMessage(Message message) throws MessagingException {
+ // Wait until the unacked state is reflected in the admin
+ QueueInfo info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
+ int n = 0;
+ while (n++ < 100 && info.getMessagesUnacknowledged() < 1L) {
+ try {
+ Thread.sleep(100);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
+ }
+ throw new RuntimeException("foo");
+ }
+
+ });
+ Binding consumerBinding = binder.bindConsumer("dlqTestManual",
+ "default", moduleInputChannel, consumerProperties);
+
+ RabbitTemplate template = new RabbitTemplate(
+ this.rabbitAvailableRule.getResource());
+ template.convertAndSend("", TEST_PREFIX + "dlqTestManual.default", "foo");
+
+ int n = 0;
+ while (n++ < 100) {
+ Object deadLetter = template
+ .receiveAndConvert(TEST_PREFIX + "dlqTestManual.default.dlq");
+ if (deadLetter != null) {
+ assertThat(deadLetter).isEqualTo("foo");
+ break;
+ }
+ Thread.sleep(100);
+ }
+ assertThat(n).isLessThan(100);
+
+ n = 0;
+ QueueInfo info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
+ while (n++ < 100 && info.getMessagesUnacknowledged() > 0L) {
+ Thread.sleep(100);
+ info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
+ }
+ assertThat(info.getMessagesUnacknowledged()).isEqualTo(0L);
+
+ consumerBinding.unbind();
+
+ ApplicationContext context = TestUtils.getPropertyValue(binder,
+ "binder.provisioningProvider.autoDeclareContext",
+ ApplicationContext.class);
+ assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.binding"))
+ .isFalse();
+ assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default")).isFalse();
+ assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.dlq.binding"))
+ .isFalse();
+ assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.dlq")).isFalse();
+ }
+
+ @Test
+ public void testAutoBindDLQPartionedConsumerFirst(TestInfo testInfo) throws Exception {
+ RabbitTestBinder binder = getBinder();
+ ExtendedConsumerProperties