GH-325: Initial Support for RabbitMQ Stream Plugin

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/325

* Do not propagate `streamContext` header on output.

* Fix test broken by previous commit.

* Remove overrides from POMs.

docs/src/main/asciidoc/overview.adoc

@@ -164,6 +164,7 @@ Default: none - the broker will generate random consumer tags.
 containerType::
 Select the type of listener container to be used.
 See https://docs.spring.io/spring-amqp/reference/html/_reference.html#choose-container[Choosing a Container] in the Spring AMQP documentation for more information.
+Also see <<rabbitmq-stream>>.
 +
 Default: `simple`
 deadLetterQueueName::
@@ -413,6 +414,62 @@ Not supported when the `containerType` is `direct`.
 +
 Default: `1`.
 
+[[rabbitmq-stream]]
+=== Initial Support for the RabbitMQ Stream Plugin
+
+Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided.
+To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`.
+
+IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is also not supported at this time.
+Only a single stream queue can be consumed by each binding.
+
+To configure the binder to use `containerType=stream`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the listener container.
+
+====
+[source, java]
+----
+@Bean
+Environment streamEnv() {
+    return Environment.builder()
+            .build();
+}
+
+@Bean
+ListenerContainerCustomizer<MessageListenerContainer> customizer() {
+    return (cont, dest, group) -> {
+        StreamListenerContainer container = (StreamListenerContainer) cont;
+        container.setConsumerCustomizer(builder -> {
+            builder.offset(OffsetSpecification.first());
+        });
+        // ...
+    };
+}
+----
+====
+
+The stream name (for the purpose of offset tracking) is set to the binding `destination + '.' + group`.
+If you decide to use manual offset tracking, the `Context` is available as a message header:
+
+====
+[source, java]
+----
+int count;
+
+@Bean
+public Consumer<Message<?>> input() {
+    return msg -> {
+        System.out.println(msg);
+        if (++count % 1000 == 0) {
+            Context context = msg.getHeaders().get("rabbitmq_streamContext", Context.class);
+            context.consumer().store(context.offset());
+        }
+    };
+}
+----
+====
+
+Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and consumer builder.
+
 === Advanced Listener Container Configuration
 
 To set listener container properties that are not exposed as binder or binding properties, add a single bean of type `ListenerContainerCustomizer` to the application context.

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java

@@ -20,7 +20,6 @@ import javax.validation.constraints.Min;
 
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.MessageDeliveryMode;
-import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType;
 import org.springframework.util.Assert;
 
 /**
@@ -348,4 +347,30 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 		this.receiveTimeout = receiveTimeout;
 	}
 
+	/**
+	 * Container type.
+	 * @author Gary Russell
+	 * @since 3.2
+	 *
+	 */
+	public enum ContainerType {
+
+		/**
+		 * Container where the RabbitMQ consumer dispatches messages to an invoker thread.
+		 */
+		SIMPLE,
+
+		/**
+		 * Container where the listener is invoked directly on the RabbitMQ consumer
+		 * thread.
+		 */
+		DIRECT,
+
+		/**
+		 * Container that uses the RabbitMQ Stream Client.
+		 */
+		STREAM
+
+	}
+
 }

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

@@ -51,6 +51,7 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties;
 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.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
@@ -268,6 +269,9 @@ public class RabbitExchangeQueueProvisioner
 		}
 		Binding binding = null;
 		if (properties.getExtension().isBindQueue()) {
+			if (properties.getExtension().getContainerType().equals(ContainerType.STREAM)) {
+				queue.getArguments().put("x-queue-type", "stream");
+			}
 			declareQueue(queueName, queue);
 			String[] routingKeys = bindingRoutingKeys(properties.getExtension());
 			if (ObjectUtils.isEmpty(routingKeys)) {

spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java

@@ -27,6 +27,9 @@ import java.util.concurrent.Executors;
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
 
@@ -72,6 +75,8 @@ public class RabbitTestSupport
 	 */
 	public static class RabbitProxy {
 
+		private static final Log LOGGER = LogFactory.getLog(RabbitProxy.class);
+
 		private final int port;
 
 		private final ExecutorService serverExec = Executors.newSingleThreadExecutor();
@@ -93,7 +98,8 @@ public class RabbitTestSupport
 
 		public void start() throws IOException {
 			this.serverSocket = ServerSocketFactory.getDefault()
-					.createServerSocket(this.port);
+					.createServerSocket(this.port, 10);
+			LOGGER.info("Proxy started");
 			this.serverExec.execute(new Runnable() {
 
 				@Override
@@ -101,6 +107,7 @@ public class RabbitTestSupport
 					try {
 						while (true) {
 							final Socket socket = serverSocket.accept();
+							LOGGER.info("Accepted Connection");
 							socketExec.execute(new Runnable() {
 
 								@Override
@@ -113,6 +120,7 @@ public class RabbitTestSupport
 
 											@Override
 											public void run() {
+												LOGGER.info("Running: " + rabbitSocket.getLocalPort());
 												try {
 													InputStream is = rabbitSocket
 															.getInputStream();

spring-cloud-stream-binder-rabbit/pom.xml

@@ -55,6 +55,12 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-amqp</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.amqp</groupId>
+			<artifactId>spring-rabbit-stream</artifactId>
+			<version>2.4.0-M1</version>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-jmx</artifactId>

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

@@ -52,6 +52,7 @@ import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
 import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
@@ -66,7 +67,6 @@ import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPos
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
-import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.BinderHeaders;
@@ -78,6 +78,7 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.HeaderMode;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties;
 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.RabbitExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
 import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
@@ -186,7 +187,7 @@ public class RabbitMessageChannelBinder extends
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
 			RabbitProperties rabbitProperties,
 			RabbitExchangeQueueProvisioner provisioningProvider,
-			ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer) {
+			ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer) {
 
 		this(connectionFactory, rabbitProperties, provisioningProvider, containerCustomizer, null);
 	}
@@ -194,7 +195,7 @@ public class RabbitMessageChannelBinder extends
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
 			RabbitProperties rabbitProperties,
 			RabbitExchangeQueueProvisioner provisioningProvider,
-			ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer,
+			ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer,
 			MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer) {
 
 		super(new String[0], provisioningProvider, containerCustomizer, sourceCustomizer);
@@ -361,6 +362,7 @@ public class RabbitMessageChannelBinder extends
 		headerPatterns.add("!" + BinderHeaders.PARTITION_HEADER);
 		headerPatterns.add("!" + IntegrationMessageHeaderAccessor.SOURCE_DATA);
 		headerPatterns.add("!" + IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT);
+		headerPatterns.add("!rabbitmq_streamContext");
 		headerPatterns.addAll(Arrays.asList(extendedProperties.getHeaderPatterns()));
 		mapper.setRequestHeaderNames(
 				headerPatterns.toArray(new String[headerPatterns.size()]));
@@ -460,6 +462,50 @@ public class RabbitMessageChannelBinder extends
 				"the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively");
 		String destination = consumerDestination.getName();
 		RabbitConsumerProperties extension = properties.getExtension();
+		MessageListenerContainer listenerContainer = createAndConfigureContainer(consumerDestination, group,
+				properties, destination, extension);
+		String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true);
+		listenerContainer.setQueueNames(queues);
+		getContainerCustomizer().configure(listenerContainer,
+				consumerDestination.getName(), group);
+		listenerContainer.afterPropertiesSet();
+
+		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
+		adapter.setBindSourceMessage(true);
+		adapter.setBeanFactory(this.getBeanFactory());
+		adapter.setBeanName("inbound." + destination);
+		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
+		mapper.setRequestHeaderNames(extension.getHeaderPatterns());
+		adapter.setHeaderMapper(mapper);
+		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(
+				consumerDestination, group, properties);
+		if (properties.getMaxAttempts() > 1) {
+			adapter.setRetryTemplate(buildRetryTemplate(properties));
+			adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
+		}
+		else {
+			adapter.setErrorMessageStrategy(errorMessageStrategy);
+			adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
+		}
+		adapter.setMessageConverter(passThoughConverter);
+		if (properties.isBatchMode() && extension.isEnableBatching()
+				&& ContainerType.SIMPLE.equals(extension.getContainerType())) {
+			adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS);
+		}
+		if (extension.getContainerType().equals(ContainerType.STREAM)) {
+			StreamContainerUtils.configureAdapter(adapter);
+		}
+		return adapter;
+	}
+
+	private MessageListenerContainer createAndConfigureContainer(ConsumerDestination consumerDestination,
+			String group, ExtendedConsumerProperties<RabbitConsumerProperties> properties, String destination,
+			RabbitConsumerProperties extension) {
+
+		if (extension.getContainerType().equals(ContainerType.STREAM)) {
+			return StreamContainerUtils.createContainer(consumerDestination, group, properties, destination, extension,
+					getApplicationContext());
+		}
 		boolean directContainer = extension.getContainerType()
 				.equals(ContainerType.DIRECT);
 		AbstractMessageListenerContainer listenerContainer = directContainer
@@ -485,8 +531,6 @@ public class RabbitMessageChannelBinder extends
 				.setRecoveryInterval(extension.getRecoveryInterval());
 		listenerContainer.setTaskExecutor(
 				new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
-		String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true);
-		listenerContainer.setQueueNames(queues);
 		listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
 		listenerContainer.setMessagePropertiesConverter(
 				RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
@@ -504,40 +548,13 @@ public class RabbitMessageChannelBinder extends
 		else if (getApplicationContext() != null) {
 			listenerContainer.setApplicationEventPublisher(getApplicationContext());
 		}
-		getContainerCustomizer().configure(listenerContainer,
-				consumerDestination.getName(), group);
 		if (StringUtils.hasText(extension.getConsumerTagPrefix())) {
 			final AtomicInteger index = new AtomicInteger();
 			listenerContainer.setConsumerTagStrategy(
 					q -> extension.getConsumerTagPrefix() + "#"
 							+ index.getAndIncrement());
 		}
-		listenerContainer.afterPropertiesSet();
-
-		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(
-				listenerContainer);
-		adapter.setBindSourceMessage(true);
-		adapter.setBeanFactory(this.getBeanFactory());
-		adapter.setBeanName("inbound." + destination);
-		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
-		mapper.setRequestHeaderNames(extension.getHeaderPatterns());
-		adapter.setHeaderMapper(mapper);
-		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(
-				consumerDestination, group, properties);
-		if (properties.getMaxAttempts() > 1) {
-			adapter.setRetryTemplate(buildRetryTemplate(properties));
-			adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
-		}
-		else {
-			adapter.setErrorMessageStrategy(errorMessageStrategy);
-			adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
-		}
-		adapter.setMessageConverter(passThoughConverter);
-		if (properties.isBatchMode() && extension.isEnableBatching()
-				&& ContainerType.SIMPLE.equals(extension.getContainerType())) {
-			adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS);
-		}
-		return adapter;
+		return listenerContainer;
 	}
 
 	private void setSMLCProperties(

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamContainerUtils.java

@@ -0,0 +1,257 @@
+/*
+ * 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;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import com.rabbitmq.stream.Codec;
+import com.rabbitmq.stream.ConsumerBuilder;
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.MessageBuilder;
+import com.rabbitmq.stream.MessageBuilder.ApplicationPropertiesBuilder;
+import com.rabbitmq.stream.MessageBuilder.PropertiesBuilder;
+import com.rabbitmq.stream.Properties;
+import com.rabbitmq.stream.codec.WrapperMessageBuilder;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+import org.springframework.rabbit.stream.support.StreamMessageProperties;
+import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
+import org.springframework.util.Assert;
+
+/**
+ * Utilities for stream containers. Used to prevent a hard runtime dependency on
+ * spring-rabbit-stream.
+ *
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+public final class StreamContainerUtils {
+
+	private StreamContainerUtils() {
+	}
+
+	/**
+	 * Create a {@link StreamListenerContainer}.
+	 *
+	 * @param consumerDestination the destination.
+	 * @param group the group.
+	 * @param properties the properties.
+	 * @param destination the destination.
+	 * @param extension the properties extension.
+	 * @param applicationContext the application context.
+	 * @return the container.
+	 */
+	public static MessageListenerContainer createContainer(ConsumerDestination consumerDestination, String group,
+			ExtendedConsumerProperties<RabbitConsumerProperties> properties, String destination,
+			RabbitConsumerProperties extension, AbstractApplicationContext applicationContext) {
+
+		StreamListenerContainer container = new StreamListenerContainer(applicationContext.getBean(Environment.class)) {
+
+			@Override
+			public synchronized void setConsumerCustomizer(Consumer<ConsumerBuilder> consumerCustomizer) {
+				super.setConsumerCustomizer(builder -> {
+					builder.name(consumerDestination.getName());
+					consumerCustomizer.accept(builder);
+				});
+			}
+
+
+		};
+		container.setMessageConverter(new DefaultStreamMessageConverter());
+		return container;
+	}
+
+	/**
+	 * Configure the channel adapter for streams support.
+	 * @param adapter the adapter.
+	 */
+	public static void configureAdapter(AmqpInboundChannelAdapter adapter) {
+		adapter.setHeaderMapper(new AmqpHeaderMapper() {
+
+			AmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
+
+			@Override
+			public Map<String, Object> toHeadersFromRequest(MessageProperties source) {
+				Map<String, Object> headers = this.mapper.toHeadersFromRequest(source);
+				headers.put("rabbitmq_streamContext", ((StreamMessageProperties) source).getContext());
+				return headers;
+			}
+
+			@Override
+			public Map<String, Object> toHeadersFromReply(MessageProperties source) {
+				return null;
+			}
+
+			@Override
+			public void fromHeadersToRequest(MessageHeaders headers, MessageProperties target) {
+			}
+
+			@Override
+			public void fromHeadersToReply(MessageHeaders headers, MessageProperties target) {
+			}
+
+		});
+	}
+
+}
+
+/**
+ * Temporary work-around for a bug in spring-rabbit-stream 2.4.0-M1.
+ */
+class DefaultStreamMessageConverter implements StreamMessageConverter {
+
+	private final Supplier<MessageBuilder> builderSupplier;
+
+	private final Charset charset = StandardCharsets.UTF_8;
+
+	/**
+	 * Construct an instance using a {@link WrapperMessageBuilder}.
+	 */
+	DefaultStreamMessageConverter() {
+		this.builderSupplier = () -> new WrapperMessageBuilder();
+	}
+
+	/**
+	 * Construct an instance using the provided codec.
+	 * @param codec the codec.
+	 */
+	DefaultStreamMessageConverter(@Nullable Codec codec) {
+		this.builderSupplier = () -> codec.messageBuilder();
+	}
+
+	@Override
+	public Message toMessage(Object object, StreamMessageProperties 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 org.springframework.amqp.core.MessageBuilder.withBody(streamMessage.getBodyAsBinary())
+				.andProperties(messageProperties)
+				.build();
+	}
+
+	@Override
+	public com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException {
+		MessageBuilder builder = this.builderSupplier.get();
+		PropertiesBuilder propsBuilder = builder.properties();
+		MessageProperties props = message.getMessageProperties();
+		Assert.isInstanceOf(StreamMessageProperties.class, props);
+		StreamMessageProperties mProps = (StreamMessageProperties) props;
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(mProps.getMessageId(), propsBuilder::messageId) // TODO different types
+				.acceptIfNotNull(mProps.getUserId(), usr -> propsBuilder.userId(usr.getBytes(this.charset)))
+				.acceptIfNotNull(mProps.getTo(), propsBuilder::to)
+				.acceptIfNotNull(mProps.getSubject(), propsBuilder::subject)
+				.acceptIfNotNull(mProps.getReplyTo(), propsBuilder::replyTo)
+				.acceptIfNotNull(mProps.getCorrelationId(), propsBuilder::correlationId) // TODO different types
+				.acceptIfNotNull(mProps.getContentType(), propsBuilder::contentType)
+				.acceptIfNotNull(mProps.getContentEncoding(), propsBuilder::contentEncoding)
+				.acceptIfNotNull(mProps.getExpiration(), exp -> propsBuilder.absoluteExpiryTime(Long.parseLong(exp)))
+				.acceptIfNotNull(mProps.getCreationTime(), propsBuilder::creationTime)
+				.acceptIfNotNull(mProps.getGroupId(), propsBuilder::groupId)
+				.acceptIfNotNull(mProps.getGroupSequence(), propsBuilder::groupSequence)
+				.acceptIfNotNull(mProps.getReplyToGroupId(), propsBuilder::replyToGroupId);
+		if (mProps.getHeaders().size() > 0) {
+			ApplicationPropertiesBuilder appPropsBuilder = builder.applicationProperties();
+			mProps.getHeaders().forEach((key, val) -> {
+				mapProp(key, val, appPropsBuilder);
+			});
+		}
+		builder.addData(message.getBody());
+		return builder.build();
+	}
+
+	private void mapProp(String key, Object val, ApplicationPropertiesBuilder builder) { // NOSONAR - complexity
+		if (val instanceof String) {
+			builder.entry(key, (String) val);
+		}
+		else if (val instanceof Long) {
+			builder.entry(key, (Long) val);
+		}
+		else if (val instanceof Integer) {
+			builder.entry(key, (Integer) val);
+		}
+		else if (val instanceof Short) {
+			builder.entry(key, (Short) val);
+		}
+		else if (val instanceof Byte) {
+			builder.entry(key, (Byte) val);
+		}
+		else if (val instanceof Double) {
+			builder.entry(key, (Double) val);
+		}
+		else if (val instanceof Float) {
+			builder.entry(key, (Float) val);
+		}
+		else if (val instanceof Character) {
+			builder.entry(key, (Character) val);
+		}
+		else if (val instanceof UUID) {
+			builder.entry(key, (UUID) val);
+		}
+		else if (val instanceof byte[]) {
+			builder.entry(key, (byte[]) val);
+		}
+	}
+
+	private void toMessageProperties(com.rabbitmq.stream.Message streamMessage,
+			StreamMessageProperties mProps) {
+
+		Properties properties = streamMessage.getProperties();
+		if (properties != null) {
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(properties.getMessageIdAsString(), mProps::setMessageId)
+					.acceptIfNotNull(properties.getUserId(), usr -> mProps.setUserId(new String(usr, this.charset)))
+					.acceptIfNotNull(properties.getTo(), mProps::setTo)
+					.acceptIfNotNull(properties.getSubject(), mProps::setSubject)
+					.acceptIfNotNull(properties.getReplyTo(), mProps::setReplyTo)
+					.acceptIfNotNull(properties.getCorrelationIdAsString(), mProps::setCorrelationId)
+					.acceptIfNotNull(properties.getContentType(), mProps::setContentType)
+					.acceptIfNotNull(properties.getContentEncoding(), mProps::setContentEncoding)
+					.acceptIfNotNull(properties.getAbsoluteExpiryTime(),
+							exp -> mProps.setExpiration(Long.toString(exp)))
+					.acceptIfNotNull(properties.getCreationTime(), mProps::setCreationTime)
+					.acceptIfNotNull(properties.getGroupId(), mProps::setGroupId)
+					.acceptIfNotNull(properties.getGroupSequence(), mProps::setGroupSequence)
+					.acceptIfNotNull(properties.getReplyToGroupId(), mProps::setReplyToGroupId);
+		}
+		Map<String, Object> applicationProperties = streamMessage.getApplicationProperties();
+		if (applicationProperties != null) {
+			mProps.getHeaders().putAll(applicationProperties);
+		}
+	}
+
+}

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java

@@ -24,7 +24,7 @@ import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy;
-import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -77,7 +77,7 @@ public class RabbitMessageChannelBinderConfiguration {
 
 	@Bean
 	RabbitMessageChannelBinder rabbitMessageChannelBinder(
-			@Nullable ListenerContainerCustomizer<AbstractMessageListenerContainer> listenerContainerCustomizer,
+			@Nullable ListenerContainerCustomizer<MessageListenerContainer> listenerContainerCustomizer,
 			@Nullable MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer,
 			@Nullable ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> producerMessageHandlerCustomizer,
 			@Nullable ConsumerEndpointCustomizer<AmqpInboundChannelAdapter> consumerCustomizer,

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

@@ -83,7 +83,6 @@ 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.boot.autoconfigure.amqp.RabbitProperties.ContainerType;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
@@ -99,6 +98,7 @@ 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;
@@ -801,7 +801,7 @@ public class RabbitBinderTests extends
 		assertThat(mode).isEqualTo(MessageDeliveryMode.PERSISTENT);
 		List<?> requestHeaders = TestUtils.getPropertyValue(endpoint,
 				"headerMapper.requestHeaderMatcher.matchers", List.class);
-		assertThat(requestHeaders).hasSize(4);
+		assertThat(requestHeaders).hasSize(5);
 		producerBinding.unbind();
 		assertThat(endpoint.isRunning()).isFalse();
 		assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional",
@@ -2340,8 +2340,8 @@ public class RabbitBinderTests extends
 	private void verifyFooRequestProducer(Lifecycle endpoint) {
 		List<?> requestMatchers = TestUtils.getPropertyValue(endpoint,
 				"headerMapper.requestHeaderMatcher.matchers", List.class);
-		assertThat(requestMatchers).hasSize(4);
-		assertThat(TestUtils.getPropertyValue(requestMatchers.get(3), "pattern"))
+		assertThat(requestMatchers).hasSize(5);
+		assertThat(TestUtils.getPropertyValue(requestMatchers.get(4), "pattern"))
 				.isEqualTo("foo");
 	}
 

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java

@@ -39,6 +39,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
@@ -381,8 +382,8 @@ public class RabbitBinderModuleTests {
 	public static class SimpleProcessor {
 
 		@Bean
-		public ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer() {
-			return (c, q, g) -> c.setBeanName(
+		public ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
+			return (c, q, g) -> ((AbstractMessageListenerContainer) c).setBeanName(
 					"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
 		}
 

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java

@@ -0,0 +1,97 @@
+/*
+ * 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.stream;
+
+import com.rabbitmq.stream.ConsumerBuilder;
+import com.rabbitmq.stream.Environment;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.binder.BinderFactory;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType;
+import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+/**
+ * @author Gary Russell
+ */
+public class RabbitStreamBinderModuleTests {
+
+	private ConfigurableApplicationContext context;
+
+	@AfterEach
+	public void tearDown() {
+		if (context != null) {
+			context.close();
+			context = null;
+		}
+	}
+
+	@Test
+	public void testExtendedProperties() {
+		context = new SpringApplicationBuilder(SimpleProcessor.class)
+				.web(WebApplicationType.NONE)
+				.run("--server.port=0");
+		BinderFactory binderFactory = context.getBean(BinderFactory.class);
+		RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
+				MessageChannel.class);
+		RabbitConsumerProperties rProps = new RabbitConsumerProperties();
+		rProps.setContainerType(ContainerType.STREAM);
+		ExtendedConsumerProperties<RabbitConsumerProperties> props =
+				new ExtendedConsumerProperties<RabbitConsumerProperties>(rProps);
+		props.setAutoStartup(false);
+		Binding<MessageChannel> binding = rabbitBinder.bindConsumer("testStream", "grp", new QueueChannel(), props);
+		assertThat(TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"))
+				.isInstanceOf(StreamListenerContainer.class);
+	}
+
+	@SpringBootApplication
+	public static class SimpleProcessor {
+
+		@Bean
+		public ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
+			return (c, q, g) -> ((StreamListenerContainer) c).setBeanName(
+					"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
+		}
+
+		@Bean
+		Environment env() {
+			Environment env = mock(Environment.class);
+			given(env.consumerBuilder()).willReturn(mock(ConsumerBuilder.class));
+			return env;
+		}
+
+	}
+
+}