GH-1352 RabbitMQ Stream Support - Initial Commit

See https://rabbitmq.github.io/rabbitmq-stream-java-client/snapshot/htmlsingle/

Basic `@RabbitListener` support.

Required a higher level abstraction for listener container factories.

* Address more PR review comments.

* Remove unnecessary `destroyMethod` property from test.

build.gradle

@@ -55,6 +55,8 @@ ext {
 	logbackVersion = '1.2.3'
 	micrometerVersion = '1.7.0'
 	mockitoVersion = '3.9.0'
+	protonJVersion = '0.33.8'
+	rabbitmqStreamVersion = '0.1.0-SNAPSHOT'
 	rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.12.0'
 	rabbitmqHttpClientVersion = '3.9.0.RELEASE'
 	reactorVersion = '2020.0.7'
@@ -96,6 +98,7 @@ allprojects {
 		maven { url 'https://repo.spring.io/libs-milestone' }
 		if (version.endsWith('-SNAPSHOT')) {
 			maven { url 'https://repo.spring.io/libs-snapshot' }
+			maven { url 'https://oss.sonatype.org/content/repositories/snapshots' }
 		}
 //		maven { url 'https://repo.spring.io/libs-staging-local' }
 	}
@@ -385,6 +388,28 @@ project('spring-rabbit') {
 
 }
 
+project('spring-rabbit-stream') {
+	description = 'Spring RabbitMQ Stream Support'
+
+	dependencies {
+
+		api project(':spring-rabbit')
+		api "com.rabbitmq:stream-client:$rabbitmqStreamVersion"
+		optionalApi "com.rabbitmq:http-client:$rabbitmqHttpClientVersion"
+
+		testApi project(':spring-rabbit-junit')
+		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core'
+		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
+		testRuntimeOnly 'com.fasterxml.jackson.dataformat:jackson-dataformat-xml'
+		testRuntimeOnly 'com.fasterxml.jackson.module:jackson-module-kotlin'
+		testRuntimeOnly "org.apache.httpcomponents:httpclient:$commonsHttpClientVersion"
+		testRuntimeOnly "org.apache.qpid:proton-j:$protonJVersion"
+		testImplementation "org.testcontainers:rabbitmq:1.15.3"
+		testImplementation "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
+	}
+
+}
+
 project('spring-rabbit-junit') {
 	description = 'Spring Rabbit JUnit Support'
 

settings.gradle

@@ -2,5 +2,6 @@ rootProject.name = 'spring-amqp-dist'
 
 include 'spring-amqp'
 include 'spring-rabbit'
+include 'spring-rabbit-stream'
 include 'spring-rabbit-junit'
 include 'spring-rabbit-test'

spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java

@@ -267,6 +267,15 @@ public final class QueueBuilder extends AbstractBuilder {
 		return withArgument("x-queue-type", "quorum");
 	}
 
+	/**
+	 * Set the queue argument to declare a queue of type 'stream' instead of 'classic'.
+	 * @return the builder.
+	 * @since 2.4
+	 */
+	public QueueBuilder stream() {
+		return withArgument("x-queue-type", "stream");
+	}
+
 	/**
 	 * Set the delivery limit; only applies to quorum queues.
 	 * @param limit the limit.

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 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.rabbit.stream.config;
+
+import java.lang.reflect.Method;
+import java.util.function.Consumer;
+
+import org.springframework.amqp.rabbit.batch.BatchingStrategy;
+import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.config.ContainerCustomizer;
+import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
+import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.lang.Nullable;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+import org.springframework.rabbit.stream.listener.adapter.StreamMessageListenerAdapter;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.stream.ConsumerBuilder;
+import com.rabbitmq.stream.Environment;
+
+/**
+ * Factory for StreamListenerContainer.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public class StreamRabbitListenerContainerFactory
+		extends BaseRabbitListenerContainerFactory<StreamListenerContainer> {
+
+	private final Environment environment;
+
+	private boolean nativeListener;
+
+	private Consumer<ConsumerBuilder> consumerCustomizer;
+
+	private ContainerCustomizer<StreamListenerContainer> containerCustomizer;
+
+	/**
+	 * Construct an instance using the provided environment.
+	 * @param environment the environment.
+	 */
+	public StreamRabbitListenerContainerFactory(Environment environment) {
+		Assert.notNull(environment, "'environment' cannot be null");
+		this.environment = environment;
+	}
+
+	/**
+	 * Set to true to create a container supporting a native RabbitMQ Stream message.
+	 * @param nativeListener true for native listeners.
+	 */
+	public void setNativeListener(boolean nativeListener) {
+		this.nativeListener = nativeListener;
+	}
+
+	/**
+	 * Customize the consumer builder before it is built.
+	 * @param consumerCustomizer the customizer.
+	 */
+	public void setConsumerCustomizer(java.util.function.Consumer<ConsumerBuilder> consumerCustomizer) {
+		this.consumerCustomizer = consumerCustomizer;
+	}
+
+	/**
+	 * Set a {@link ContainerCustomizer} that is invoked after a container is created and
+	 * configured to enable further customization of the container.
+	 * @param containerCustomizer the customizer.
+	 */
+	public void setContainerCustomizer(ContainerCustomizer<StreamListenerContainer> containerCustomizer) {
+		this.containerCustomizer = containerCustomizer;
+	}
+
+	@Override
+	public StreamListenerContainer createListenerContainer(RabbitListenerEndpoint endpoint) {
+		if (endpoint instanceof MethodRabbitListenerEndpoint && this.nativeListener) {
+			((MethodRabbitListenerEndpoint) endpoint).setAdapterProvider(
+					(boolean batch, Object bean, Method method, boolean returnExceptions,
+							RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy) -> {
+
+								Assert.isTrue(!batch, "Batch listeners are not supported by the stream container");
+								return new StreamMessageListenerAdapter(bean, method, returnExceptions, errorHandler);
+							});
+		}
+		StreamListenerContainer container = createContainerInstance();
+		if (this.consumerCustomizer != null) {
+			container.setConsumerCustomizer(this.consumerCustomizer);
+		}
+		applyCommonOverrides(endpoint, container);
+		if (this.containerCustomizer != null) {
+			this.containerCustomizer.configure(container);
+		}
+		return container;
+	}
+
+	/**
+	 * Create an instance of the listener container.
+	 * @return the container.
+	 */
+	protected StreamListenerContainer createContainerInstance() {
+		return new StreamListenerContainer(this.environment);
+	}
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java

@@ -0,0 +1,201 @@
+/*
+ * Copyright 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.rabbit.stream.listener;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.lang.Nullable;
+import org.springframework.rabbit.stream.support.converter.DefaultStreamMessageConverter;
+import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.stream.Consumer;
+import com.rabbitmq.stream.ConsumerBuilder;
+import com.rabbitmq.stream.Environment;
+
+/**
+ * A listener container for RabbitMQ Streams.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public class StreamListenerContainer implements MessageListenerContainer, BeanNameAware {
+
+	protected Log logger = LogFactory.getLog(getClass());
+
+	private final Environment environment;
+
+	private final ConsumerBuilder builder;
+
+	private StreamMessageConverter messageConverter;
+
+	private java.util.function.Consumer<ConsumerBuilder> consumerCustomizer = c -> { };
+
+	private String stream;
+
+	private Consumer consumer;
+
+	private String listenerId;
+
+	private String beanName;
+
+	private boolean autoStartup = true;
+
+	private MessageListener messageListener;
+
+	/**
+	 * Construct an instance using the provided environment.
+	 * @param environment the environment.
+	 */
+	public StreamListenerContainer(Environment environment) {
+		Assert.notNull(environment, "'environment' cannot be null");
+		this.environment = environment;
+		this.builder = environment.consumerBuilder();
+		this.messageConverter = new DefaultStreamMessageConverter(environment);
+	}
+
+	@Override
+	public void setQueueNames(String... queueNames) {
+		Assert.isTrue(queueNames != null && queueNames.length == 1, "Only one stream is supported");
+		this.stream = queueNames[0];
+		this.builder.stream(this.stream);
+	}
+
+	/**
+	 * Get a {@link StreamMessageConverter} used to convert a
+	 * {@link com.rabbitmq.stream.Message} to a
+	 * {@link org.springframework.amqp.core.Message}.
+	 * @return the converter.
+	 */
+	public StreamMessageConverter getMessageConverter() {
+		return this.messageConverter;
+	}
+
+	/**
+	 * Set a {@link StreamMessageConverter} used to convert a
+	 * {@link com.rabbitmq.stream.Message} to a
+	 * {@link org.springframework.amqp.core.Message}.
+	 * @param messageConverter the converter.
+	 */
+	public void setMessageConverter(StreamMessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+	}
+
+	/**
+	 * Customize the consumer builder before it is built.
+	 * @param consumerCustomizer the customizer.
+	 */
+	public void setConsumerCustomizer(java.util.function.Consumer<ConsumerBuilder> consumerCustomizer) {
+		this.consumerCustomizer = consumerCustomizer;
+	}
+
+	/**
+	 * The 'id' attribute of the listener.
+	 * @return the id (or the container bean name if no id set).
+	 */
+	@Nullable
+	public String getListenerId() {
+		return this.listenerId != null ? this.listenerId : this.beanName;
+	}
+
+	@Override
+	public void setListenerId(String listenerId) {
+		this.listenerId = listenerId;
+	}
+
+	/**
+	 * Return the bean name.
+	 * @return the bean name.
+	 */
+	@Nullable
+	public String getBeanName() {
+		return this.beanName;
+	}
+
+	@Override
+	public void setBeanName(String beanName) {
+		this.beanName = beanName;
+	}
+
+	@Override
+	public void setAutoStartup(boolean autoStart) {
+		this.autoStartup = autoStart;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+	@Override
+	@Nullable
+	public Object getMessageListener() {
+		return this.messageListener;
+	}
+
+	@Override
+	public synchronized boolean isRunning() {
+		return this.consumer != null;
+	}
+
+	@Override
+	public synchronized void start() {
+		if (this.consumer == null) {
+			this.consumerCustomizer.accept(this.builder);
+			this.consumer = this.builder.build();
+		}
+	}
+
+	@Override
+	public synchronized void stop() {
+		if (this.consumer != null) {
+			this.consumer.close();
+			this.consumer = null;
+		}
+	}
+
+	@Override
+	public void setupMessageListener(MessageListener messageListener) {
+		this.messageListener = messageListener;
+		this.builder.messageHandler((context, message) -> {
+			if (messageListener instanceof StreamMessageListener) {
+				((StreamMessageListener) messageListener).onStreamMessage(message, context);
+			}
+			else {
+				Message message2 = this.messageConverter.toMessage(message, new StreamMessageProperties(context));
+				if (messageListener instanceof ChannelAwareMessageListener) {
+					try {
+						((ChannelAwareMessageListener) messageListener).onMessage(message2, null);
+					}
+					catch (Exception e) { // NOSONAR
+						this.logger.error("Listner threw an exception", e);
+					}
+				}
+				else {
+					messageListener.onMessage(message2);
+				}
+			}
+		});
+	}
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageListener.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 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.rabbit.stream.listener;
+
+import org.springframework.amqp.core.MessageListener;
+
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * A message listener that receives native stream messages.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public interface StreamMessageListener extends MessageListener {
+
+	/**
+	 * Process a message.
+	 * @param message the message.
+	 * @param context the stream context.
+	 */
+	void onStreamMessage(Message message, Context context);
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageProperties.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 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.rabbit.stream.listener;
+
+import org.springframework.amqp.core.MessageProperties;
+
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * {@link MessageProperties} extension for stream messages.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public class StreamMessageProperties extends MessageProperties {
+
+	private static final long serialVersionUID = 1L;
+
+	private final Context context;
+
+	/**
+	 * Create a new instance with the provided context.
+	 * @param context the context.
+	 */
+	public StreamMessageProperties(Context context) {
+		this.context = context;
+	}
+
+	/**
+	 * Return the stream {@link Context} for the message.
+	 * @return the context.
+	 */
+	public Context getContext() {
+		return this.context;
+	}
+
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/adapter/StreamMessageListenerAdapter.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 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.rabbit.stream.listener.adapter;
+
+import java.lang.reflect.Method;
+
+import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
+import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
+import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.rabbit.stream.listener.StreamMessageListener;
+
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * A listener adapter that receives native stream messages.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public class StreamMessageListenerAdapter extends MessagingMessageListenerAdapter implements StreamMessageListener {
+
+	/**
+	 * Construct an instance with the provided arguments.
+	 * @param bean the bean.
+	 * @param method the method.
+	 * @param returnExceptions true to return exceptions.
+	 * @param errorHandler the error handler.
+	 */
+	public StreamMessageListenerAdapter(Object bean, Method method, boolean returnExceptions,
+			RabbitListenerErrorHandler errorHandler) {
+
+		super(bean, method, returnExceptions, errorHandler);
+	}
+
+	@Override
+	public void onStreamMessage(Message message, Context context) {
+		try {
+			InvocationResult result = getHandlerAdapter().invoke(null, message, context);
+			if (result.getReturnValue() != null) {
+				logger.warn("Replies are not currently supported with native Stream listeners");
+			}
+			else {
+				logger.trace("No result object given - no result to handle");
+			}
+		}
+		catch (Exception ex) {
+			this.logger.error("Failed to invoke listener", ex);
+		}
+	}
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverter.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 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.rabbit.stream.support.converter;
+
+import java.util.Map;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageBuilder;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.MessageBuilder.PropertiesBuilder;
+import com.rabbitmq.stream.Properties;
+import com.rabbitmq.stream.codec.WrapperMessageBuilder;
+
+/**
+ * Default {@link StreamMessageConverter}.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public class DefaultStreamMessageConverter implements StreamMessageConverter {
+
+	private final Environment environment;
+
+	/**
+	 * Construct an instance using the provided environment.
+	 * @param environment the environment.
+	 */
+	public DefaultStreamMessageConverter(Environment environment) {
+		Assert.notNull(environment, "'environment' cannot be null");
+		this.environment = environment;
+	}
+
+	@Override
+	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		Assert.isInstanceOf(com.rabbitmq.stream.Message.class, object);
+		com.rabbitmq.stream.Message streamMessage = (com.rabbitmq.stream.Message) object;
+		toMessageProperties(streamMessage, messageProperties);
+		return MessageBuilder.withBody(streamMessage.getBodyAsBinary()).andProperties(messageProperties).build();
+	}
+
+	@Override
+	public com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException {
+		// TODO get the builder from the environment's codec
+		WrapperMessageBuilder builder = new WrapperMessageBuilder();
+		PropertiesBuilder propsBuilder = builder.properties();
+		MessageProperties mProps = message.getMessageProperties();
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(mProps.getMessageId(), propsBuilder::messageId);
+		// TODO ...
+		builder.addData(message.getBody());
+		return builder.build();
+	}
+
+	private void toMessageProperties(com.rabbitmq.stream.Message streamMessage, MessageProperties messageProperties) {
+		Properties properties = streamMessage.getProperties();
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(properties.getMessageIdAsString(), messageProperties::setMessageId);
+		// TODO ...
+		Map<String, Object> applicationProperties = streamMessage.getApplicationProperties();
+		if (applicationProperties != null) {
+			messageProperties.getHeaders().putAll(applicationProperties);
+		}
+	}
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/StreamMessageConverter.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 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.rabbit.stream.support.converter;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.support.converter.MessageConverter;
+
+/**
+ * Converts between {@link com.rabbitmq.stream.Message} and
+ * {@link org.springframework.amqp.core.Message}.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public interface StreamMessageConverter extends MessageConverter {
+
+	@Override
+	Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException;
+
+	@Override
+	com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException;
+
+}

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/AbstractIntegrationTests.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 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.rabbit.stream.listener;
+
+import java.time.Duration;
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public abstract class AbstractIntegrationTests {
+
+	static final GenericContainer<?> RABBITMQ = new GenericContainer<>(
+			DockerImageName.parse("pivotalrabbitmq/rabbitmq-stream"))
+				.withExposedPorts(5672, 15672, 5552)
+				.withStartupTimeout(Duration.ofMinutes(2));
+
+	static {
+		RABBITMQ.start();
+	}
+
+}

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java

@@ -0,0 +1,181 @@
+/*
+ * Copyright 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.rabbit.stream.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import com.rabbitmq.http.client.Client;
+import com.rabbitmq.http.client.domain.QueueInfo;
+import com.rabbitmq.stream.Address;
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler.Context;
+import com.rabbitmq.stream.OffsetSpecification;
+
+/**
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class RabbitListenerTests extends AbstractIntegrationTests {
+
+	@Autowired
+	Config config;
+
+	@Test
+	void simple(@Autowired RabbitTemplate template) throws InterruptedException {
+
+		template.convertAndSend("test.stream.queue1", "foo");
+		assertThat(this.config.latch1.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.received).isEqualTo("foo");
+	}
+
+	@Test
+	void nativeMsg(@Autowired RabbitTemplate template) throws InterruptedException {
+
+		template.convertAndSend("test.stream.queue2", "foo");
+		assertThat(this.config.latch2.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.receivedNative).isNotNull();
+		assertThat(this.config.context).isNotNull();
+	}
+
+	@Test
+	void queueOverAmqp() throws Exception {
+		Client client = new Client("http://guest:guest@localhost:" + RABBITMQ.getMappedPort(15672) + "/api");
+		QueueInfo queue = client.getQueue("/", "stream.created.over.amqp");
+		assertThat(queue.getArguments().get("x-queue-type")).isEqualTo("stream");
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableRabbit
+	public static class Config {
+
+		final CountDownLatch latch1 = new CountDownLatch(1);
+
+		final CountDownLatch latch2 = new CountDownLatch(1);
+
+		volatile String received;
+
+		volatile Message receivedNative;
+
+		volatile Context context;
+
+		@Bean
+		Environment environment() {
+			return Environment.builder()
+					.addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552)))
+					.build();
+		}
+
+		@Bean
+		SmartLifecycle creator(Environment env) {
+			return new SmartLifecycle() {
+
+				@Override
+				public void stop() {
+				}
+
+				@Override
+				public void start() {
+					env.streamCreator().stream("test.stream.queue1").create();
+					env.streamCreator().stream("test.stream.queue2").create();
+				}
+
+				@Override
+				public boolean isRunning() {
+					return false;
+				}
+			};
+		}
+
+		@Bean
+		RabbitListenerContainerFactory<StreamListenerContainer> rabbitListenerContainerFactory(Environment env) {
+			return new StreamRabbitListenerContainerFactory(env);
+		}
+
+		@RabbitListener(queues = "test.stream.queue1")
+		void listen(String in) {
+			this.received = in;
+			this.latch1.countDown();
+		}
+
+		@Bean
+		RabbitListenerContainerFactory<StreamListenerContainer> nativeFactory(Environment env) {
+			StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
+			factory.setNativeListener(true);
+			factory.setConsumerCustomizer(builder -> builder.name("myConsumer")
+					.offset(OffsetSpecification.first())
+					.manualCommitStrategy());
+			return factory;
+		}
+
+		@RabbitListener(queues = "test.stream.queue2", containerFactory = "nativeFactory")
+		void nativeMsg(Message in, Context context) {
+			this.receivedNative = in;
+			this.context = context;
+			this.latch2.countDown();
+			context.commit();
+		}
+
+		@Bean
+		CachingConnectionFactory cf() {
+			return new CachingConnectionFactory(RABBITMQ.getContainerIpAddress(), RABBITMQ.getFirstMappedPort());
+		}
+
+		@Bean
+		RabbitTemplate template(CachingConnectionFactory cf) {
+			return new RabbitTemplate(cf);
+		}
+
+		@Bean
+		RabbitAdmin admin(CachingConnectionFactory cf) {
+			return new RabbitAdmin(cf);
+		}
+
+		@Bean
+		Queue queue() {
+			return QueueBuilder.durable("stream.created.over.amqp")
+					.stream()
+					.build();
+		}
+
+ 	}
+
+}

spring-rabbit-stream/src/test/resources/log4j2-test.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration>
+	<Appenders>
+		<Console name="STDOUT" target="SYSTEM_OUT">
+			<PatternLayout pattern="%d %5p %c [%t] : %m%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Logger name="org.springframework.amqp.rabbit" level="info"/>
+		<Logger name="org.springframework.beans.factory" level="info"/>
+		<Root level="info">
+			<AppenderRef ref="STDOUT" />
+		</Root>
+	</Loggers>
+</Configuration>

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java

@@ -32,7 +32,6 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
-import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener;
 import org.springframework.amqp.support.ConsumerTagStrategy;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.utils.JavaUtils;
@@ -42,8 +41,6 @@ import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.lang.Nullable;
-import org.springframework.retry.RecoveryCallback;
-import org.springframework.retry.support.RetryTemplate;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
@@ -64,7 +61,8 @@ import org.springframework.util.backoff.FixedBackOff;
  * @see AbstractMessageListenerContainer
  */
 public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractMessageListenerContainer>
-		implements RabbitListenerContainerFactory<C>, ApplicationContextAware, ApplicationEventPublisherAware {
+		extends BaseRabbitListenerContainerFactory<C>
+		implements ApplicationContextAware, ApplicationEventPublisherAware {
 
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
 
@@ -88,8 +86,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private Boolean globalQos;
 
-	private Boolean defaultRequeueRejected;
-
 	private Advice[] adviceChain;
 
 	private BackOff recoveryBackOff;
@@ -114,12 +110,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private MessagePostProcessor[] afterReceivePostProcessors;
 
-	private MessagePostProcessor[] beforeSendReplyPostProcessors;
-
-	private RetryTemplate retryTemplate;
-
-	private RecoveryCallback<?> recoveryCallback;
-
 	private ContainerCustomizer<C> containerCustomizer;
 
 	private boolean batchListener;
@@ -192,14 +182,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		this.prefetchCount = prefetch;
 	}
 
-	/**
-	 * @param requeueRejected true to reject by default.
-	 * @see AbstractMessageListenerContainer#setDefaultRequeueRejected
-	 */
-	public void setDefaultRequeueRejected(Boolean requeueRejected) {
-		this.defaultRequeueRejected = requeueRejected;
-	}
-
 	/**
 	 * @return the advice chain that was set. Defaults to {@code null}.
 	 * @since 1.7.4
@@ -309,44 +291,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		this.afterReceivePostProcessors = Arrays.copyOf(postProcessors, postProcessors.length);
 	}
 
-	/**
-	 * Set post processors that will be applied before sending replies; added to each
-	 * message listener adapter.
-	 * @param postProcessors the post processors.
-	 * @since 2.0.3
-	 * @see AbstractAdaptableMessageListener#setBeforeSendReplyPostProcessors(MessagePostProcessor...)
-	 */
-	public void setBeforeSendReplyPostProcessors(MessagePostProcessor... postProcessors) {
-		Assert.notNull(postProcessors, "'postProcessors' cannot be null");
-		Assert.noNullElements(postProcessors, "'postProcessors' cannot have null elements");
-		this.beforeSendReplyPostProcessors = Arrays.copyOf(postProcessors, postProcessors.length);
-	}
-
-	/**
-	 * Set a {@link RetryTemplate} to use when sending replies; added to each message
-	 * listener adapter.
-	 * @param retryTemplate the template.
-	 * @since 2.0.6
-	 * @see #setReplyRecoveryCallback(RecoveryCallback)
-	 * @see AbstractAdaptableMessageListener#setRetryTemplate(RetryTemplate)
-	 */
-	public void setRetryTemplate(RetryTemplate retryTemplate) {
-		this.retryTemplate = retryTemplate;
-	}
-
-	/**
-	 * Set a {@link RecoveryCallback} to invoke when retries are exhausted. Added to each
-	 * message listener adapter. Only used if a {@link #setRetryTemplate(RetryTemplate)
-	 * retryTemplate} is provided.
-	 * @param recoveryCallback the recovery callback.
-	 * @since 2.0.6
-	 * @see #setRetryTemplate(RetryTemplate)
-	 * @see AbstractAdaptableMessageListener#setRecoveryCallback(RecoveryCallback)
-	 */
-	public void setReplyRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
-		this.recoveryCallback = recoveryCallback;
-	}
-
 	/**
 	 * Set a {@link ContainerCustomizer} that is invoked after a container is created and
 	 * configured to enable further customization of the container.
@@ -437,29 +381,14 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		}
 		if (endpoint != null) { // endpoint settings overriding default factory settings
 			javaUtils
-				.acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup)
 				.acceptIfNotNull(endpoint.getTaskExecutor(), instance::setTaskExecutor)
-				.acceptIfNotNull(endpoint.getAckMode(), instance::setAcknowledgeMode);
-			javaUtils
+				.acceptIfNotNull(endpoint.getAckMode(), instance::setAcknowledgeMode)
 				.acceptIfNotNull(this.batchingStrategy, endpoint::setBatchingStrategy);
 			instance.setListenerId(endpoint.getId());
 			endpoint.setBatchListener(this.batchListener);
-			endpoint.setupListenerContainer(instance);
-		}
-		if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {
-			AbstractAdaptableMessageListener messageListener = (AbstractAdaptableMessageListener) instance
-					.getMessageListener();
-			javaUtils
-					.acceptIfNotNull(this.beforeSendReplyPostProcessors,
-							messageListener::setBeforeSendReplyPostProcessors)
-					.acceptIfNotNull(this.retryTemplate, messageListener::setRetryTemplate)
-					.acceptIfCondition(this.retryTemplate != null && this.recoveryCallback != null,
-							this.recoveryCallback, messageListener::setRecoveryCallback)
-					.acceptIfNotNull(this.defaultRequeueRejected, messageListener::setDefaultRequeueRejected)
-					.acceptIfNotNull(endpoint.getReplyPostProcessor(), messageListener::setReplyPostProcessor)
-					.acceptIfNotNull(endpoint.getReplyContentType(), messageListener::setReplyContentType);
-			messageListener.setConverterWinsContentType(endpoint.isConverterWinsContentType());
 		}
+		applyCommonOverrides(endpoint, instance);
+
 		initializeContainer(instance, endpoint);
 
 		if (this.containerCustomizer != null) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/BaseRabbitListenerContainerFactory.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright 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.amqp.rabbit.config;
+
+import java.util.Arrays;
+
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
+import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener;
+import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.lang.Nullable;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * Base abstract class for listener container factories.
+ *
+ * @param <C> the container type that the factory creates.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+public abstract class BaseRabbitListenerContainerFactory<C extends MessageListenerContainer>
+	implements RabbitListenerContainerFactory<C> {
+
+	protected Boolean defaultRequeueRejected;
+
+	private MessagePostProcessor[] beforeSendReplyPostProcessors;
+
+	private RetryTemplate retryTemplate;
+
+	private RecoveryCallback<?> recoveryCallback;
+
+	@Override
+	public abstract C createListenerContainer(RabbitListenerEndpoint endpoint);
+
+	/**
+	 * @param requeueRejected true to reject by default.
+	 * @see AbstractMessageListenerContainer#setDefaultRequeueRejected
+	 */
+	public void setDefaultRequeueRejected(Boolean requeueRejected) {
+		this.defaultRequeueRejected = requeueRejected;
+	}
+
+	/**
+	 * Set post processors that will be applied before sending replies; added to each
+	 * message listener adapter.
+	 * @param postProcessors the post processors.
+	 * @since 2.0.3
+	 * @see AbstractAdaptableMessageListener#setBeforeSendReplyPostProcessors(MessagePostProcessor...)
+	 */
+	public void setBeforeSendReplyPostProcessors(MessagePostProcessor... postProcessors) {
+		Assert.notNull(postProcessors, "'postProcessors' cannot be null");
+		Assert.noNullElements(postProcessors, "'postProcessors' cannot have null elements");
+		this.beforeSendReplyPostProcessors = Arrays.copyOf(postProcessors, postProcessors.length);
+	}
+
+	/**
+	 * Set a {@link RetryTemplate} to use when sending replies; added to each message
+	 * listener adapter.
+	 * @param retryTemplate the template.
+	 * @since 2.0.6
+	 * @see #setReplyRecoveryCallback(RecoveryCallback)
+	 * @see AbstractAdaptableMessageListener#setRetryTemplate(RetryTemplate)
+	 */
+	public void setRetryTemplate(RetryTemplate retryTemplate) {
+		this.retryTemplate = retryTemplate;
+	}
+
+	/**
+	 * Set a {@link RecoveryCallback} to invoke when retries are exhausted. Added to each
+	 * message listener adapter. Only used if a {@link #setRetryTemplate(RetryTemplate)
+	 * retryTemplate} is provided.
+	 * @param recoveryCallback the recovery callback.
+	 * @since 2.0.6
+	 * @see #setRetryTemplate(RetryTemplate)
+	 * @see AbstractAdaptableMessageListener#setRecoveryCallback(RecoveryCallback)
+	 */
+	public void setReplyRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
+		this.recoveryCallback = recoveryCallback;
+	}
+
+	protected void applyCommonOverrides(@Nullable RabbitListenerEndpoint endpoint, C instance) {
+		if (endpoint != null) { // endpoint settings overriding default factory settings
+			JavaUtils.INSTANCE
+				.acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup);
+			instance.setListenerId(endpoint.getId());
+			endpoint.setupListenerContainer(instance);
+		}
+		if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {
+			AbstractAdaptableMessageListener messageListener = (AbstractAdaptableMessageListener) instance
+					.getMessageListener();
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(this.beforeSendReplyPostProcessors,
+							messageListener::setBeforeSendReplyPostProcessors)
+					.acceptIfNotNull(this.retryTemplate, messageListener::setRetryTemplate)
+					.acceptIfCondition(this.retryTemplate != null && this.recoveryCallback != null,
+							this.recoveryCallback, messageListener::setRecoveryCallback)
+					.acceptIfNotNull(this.defaultRequeueRejected, messageListener::setDefaultRequeueRejected)
+					.acceptIfNotNull(endpoint.getReplyPostProcessor(), messageListener::setReplyPostProcessor)
+					.acceptIfNotNull(endpoint.getReplyContentType(), messageListener::setReplyContentType);
+			messageListener.setConverterWinsContentType(endpoint.isConverterWinsContentType());
+		}
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ContainerCustomizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-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.
@@ -16,7 +16,7 @@
 
 package org.springframework.amqp.rabbit.config;
 
-import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 
 /**
  * Called by the container factory after the container is created and configured.
@@ -28,7 +28,7 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer
  *
  */
 @FunctionalInterface
-public interface ContainerCustomizer<C extends AbstractMessageListenerContainer> {
+public interface ContainerCustomizer<C extends MessageListenerContainer> {
 
 	/**
 	 * Configure the container.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -296,6 +296,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 	 * Set the name of the queue(s) to receive messages from.
 	 * @param queueName the desired queueName(s) (can not be <code>null</code>)
 	 */
+	@Override
 	public void setQueueNames(String... queueName) {
 		Assert.noNullElements(queueName, "Queue name(s) cannot be null");
 		setQueues(Arrays.stream(queueName)
@@ -454,9 +455,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		}
 	}
 
-	/**
-	 * @return The message listener object to register.
-	 */
+	@Override
 	public Object getMessageListener() {
 		return this.messageListener;
 	}
@@ -557,6 +556,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 	 *
 	 * @param autoStartup true for auto startup.
 	 */
+	@Override
 	public void setAutoStartup(boolean autoStartup) {
 		this.autoStartup = autoStartup;
 	}
@@ -1189,7 +1189,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
 	@Override
-	public final void afterPropertiesSet() {
+	public void afterPropertiesSet() {
 		super.afterPropertiesSet();
 		Assert.state(
 				this.exposeListenerChannel || !getAcknowledgeMode().isManual(),

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-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.
@@ -372,31 +372,38 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
 
 	@Override
 	public void setupListenerContainer(MessageListenerContainer listenerContainer) {
-		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer;
+		Collection<String> qNames = getQueueNames();
+		boolean queueNamesEmpty = qNames.isEmpty();
+		if (listenerContainer instanceof AbstractMessageListenerContainer) {
+			AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer;
 
-		boolean queuesEmpty = getQueues().isEmpty();
-		boolean queueNamesEmpty = getQueueNames().isEmpty();
-		if (!queuesEmpty && !queueNamesEmpty) {
-			throw new IllegalStateException("Queues or queue names must be provided but not both for " + this);
-		}
-		if (queuesEmpty) {
-			Collection<String> names = getQueueNames();
-			container.setQueueNames(names.toArray(new String[names.size()]));
+			boolean queuesEmpty = getQueues().isEmpty();
+			if (!queuesEmpty && !queueNamesEmpty) {
+				throw new IllegalStateException("Queues or queue names must be provided but not both for " + this);
+			}
+			if (queuesEmpty) {
+				Collection<String> names = qNames;
+				container.setQueueNames(names.toArray(new String[0]));
+			}
+			else {
+				Collection<Queue> instances = getQueues();
+				container.setQueues(instances.toArray(new Queue[0]));
+			}
+
+			container.setExclusive(isExclusive());
+			if (getPriority() != null) {
+				Map<String, Object> args = container.getConsumerArguments();
+				args.put("x-priority", getPriority());
+				container.setConsumerArguments(args);
+			}
+
+			if (getAdmin() != null) {
+				container.setAmqpAdmin(getAdmin());
+			}
 		}
 		else {
-			Collection<Queue> instances = getQueues();
-			container.setQueues(instances.toArray(new Queue[instances.size()]));
-		}
-
-		container.setExclusive(isExclusive());
-		if (getPriority() != null) {
-			Map<String, Object> args = container.getConsumerArguments();
-			args.put("x-priority", getPriority());
-			container.setConsumerArguments(args);
-		}
-
-		if (getAdmin() != null) {
-			container.setAmqpAdmin(getAdmin());
+			Assert.state(!queueNamesEmpty, "At least one queue name is required");
+			listenerContainer.setQueueNames(qNames.toArray(new String[0]));
 		}
 		setupMessageListener(listenerContainer);
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-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.
@@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.listener;
 
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.context.SmartLifecycle;
+import org.springframework.lang.Nullable;
 
 /**
  * Internal abstraction used by the framework representing a message
@@ -57,4 +58,33 @@ public interface MessageListenerContainer extends SmartLifecycle {
 		return false;
 	}
 
+	/**
+	 * Set the queue names.
+	 * @param queues the queue names.
+	 * @since 2.4
+	 */
+	void setQueueNames(String... queues);
+
+	/**
+	 * Set auto startup.
+	 * @param autoStart true to auto start.
+	 * @since 2.4
+	 */
+	void setAutoStartup(boolean autoStart);
+
+	/**
+	 * Get the message listener.
+	 * @return The message listener object.
+	 * @since 2.4
+	 */
+	@Nullable
+	Object getMessageListener();
+
+	/**
+	 * Set the listener id.
+	 * @param id the id.
+	 * @since 2.4
+	 */
+	void setListenerId(String id);
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java

@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 
+import org.springframework.amqp.rabbit.batch.BatchingStrategy;
 import org.springframework.amqp.rabbit.listener.adapter.BatchMessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
@@ -53,6 +54,8 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
 
 	private RabbitListenerErrorHandler errorHandler;
 
+	private AdapterProvider adapterProvider = new DefaultAdapterProvider();
+
 	/**
 	 * Set the object instance that should manage this endpoint.
 	 * @param bean the target bean instance.
@@ -114,6 +117,15 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
 		return this.messageHandlerMethodFactory;
 	}
 
+	/**
+	 * Set a provider to create adapter instances.
+	 * @param adapterProvider the provider.
+	 */
+	public void setAdapterProvider(AdapterProvider adapterProvider) {
+		Assert.notNull(adapterProvider, "'adapterProvider' cannot be null");
+		this.adapterProvider = adapterProvider;
+	}
+
 	@Override
 	protected MessagingMessageListenerAdapter createMessageListener(MessageListenerContainer container) {
 		Assert.state(this.messageHandlerMethodFactory != null,
@@ -150,14 +162,8 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
 	 * @return the {@link MessagingMessageListenerAdapter} instance.
 	 */
 	protected MessagingMessageListenerAdapter createMessageListenerInstance() {
-		if (isBatchListener()) {
-			return new BatchMessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions,
+		return this.adapterProvider.getAdapter(isBatchListener(), this.bean, this.method, this.returnExceptions,
 					this.errorHandler, getBatchingStrategy());
-		}
-		else {
-			return new MessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions,
-					this.errorHandler);
-		}
 	}
 
 	@Nullable
@@ -196,4 +202,43 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
 				.append(" | method='").append(this.method).append("'");
 	}
 
+	/**
+	 * Provider of listener adapters.
+	 * @since 2.4
+	 *
+	 */
+	public interface AdapterProvider {
+
+		/**
+		 * Get an adapter instance.
+		 * @param batch true for a batch listener.
+		 * @param bean the bean.
+		 * @param method the method.
+		 * @param returnExceptions true to return exceptions.
+		 * @param errorHandler the error handler.
+		 * @param batchingStrategy the batching strategy for batch listeners.
+		 * @return the adapter.
+		 */
+		MessagingMessageListenerAdapter getAdapter(boolean batch, Object bean, Method method, boolean returnExceptions,
+				RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy);
+	}
+
+	private static final class DefaultAdapterProvider implements AdapterProvider {
+
+		@Override
+		public MessagingMessageListenerAdapter getAdapter(boolean batch, Object bean, Method method,
+				boolean returnExceptions, RabbitListenerErrorHandler errorHandler,
+				@Nullable BatchingStrategy batchingStrategy) {
+
+			if (batch) {
+				return new BatchMessagingMessageListenerAdapter(bean, method, returnExceptions, errorHandler,
+						batchingStrategy);
+			}
+			else {
+				return new MessagingMessageListenerAdapter(bean, method, returnExceptions, errorHandler);
+			}
+		}
+
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java

@@ -44,7 +44,9 @@ public class MultiMethodRabbitListenerEndpoint extends MethodRabbitListenerEndpo
 	 * Construct an instance for the provided methods and bean.
 	 * @param methods the methods.
 	 * @param bean the bean.
+	 * @deprecated - no longer used.
 	 */
+	@Deprecated
 	public MultiMethodRabbitListenerEndpoint(List<Method> methods, Object bean) {
 		this(methods, null, bean);
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -103,6 +103,10 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 		this.handlerAdapter = handlerAdapter;
 	}
 
+	protected HandlerAdapter getHandlerAdapter() {
+		return this.handlerAdapter;
+	}
+
 	/**
 	 * Set the {@link AmqpHeaderMapper} implementation to use to map the standard
 	 * AMQP headers. By default, a {@link org.springframework.amqp.support.SimpleAmqpHeaderMapper

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -20,6 +20,7 @@ import java.util.List;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
+import org.springframework.lang.Nullable;
 
 import com.rabbitmq.client.Channel;
 
@@ -37,10 +38,11 @@ public interface ChannelAwareMessageListener extends MessageListener {
 	 * <p>Implementors are supposed to process the given Message,
 	 * typically sending reply messages through the given Session.
 	 * @param message the received AMQP message (never <code>null</code>)
-	 * @param channel the underlying Rabbit Channel (never <code>null</code>)
+	 * @param channel the underlying Rabbit Channel (never <code>null</code>
+	 * unless called by the stream listener container).
 	 * @throws Exception Any.
 	 */
-	void onMessage(Message message, Channel channel) throws Exception; // NOSONAR
+	void onMessage(Message message, @Nullable Channel channel) throws Exception; // NOSONAR
 
 	@Override
 	default void onMessage(Message message) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MessageListenerTestContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -21,6 +21,7 @@ import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.lang.Nullable;
 
 /**
  * @author Stephane Nicoll
@@ -47,6 +48,24 @@ public class MessageListenerTestContainer
 		return endpoint;
 	}
 
+	@Override
+	public void setQueueNames(String... queues) {
+	}
+
+	@Override
+	public void setAutoStartup(boolean autoStart) {
+	}
+
+	@Override
+	@Nullable
+	public Object getMessageListener() {
+		return null;
+	}
+
+	@Override
+	public void setListenerId(String id) {
+	}
+
 	public boolean isStarted() {
 		return startInvoked && initializationInvoked;
 	}