GH-3626: RabbitMQ Stream Support (#3895)

Resolves https://github.com/spring-projects/spring-integration/issues/3626

- spec for stream listener container
- move message handler from scst to here; add spec

* Fix test.

* Polishing and Docs.

* Fix anchors in doc.

* Don't Extend `AbstractMessageProducingHandler`

build.gradle

@@ -99,7 +99,7 @@ ext {
     rsocketVersion = '1.1.3'
     servletApiVersion = '5.0.0'
     smackVersion = '4.4.6'
-    springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '3.0.0-M4'
+    springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '3.0.0-SNAPSHOT'
     springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2022.0.0-M6'
     springGraphqlVersion = '1.1.0-M1'
     springKafkaVersion = '3.0.0-M6'
@@ -470,12 +470,16 @@ project('spring-integration-amqp') {
         api("org.springframework.amqp:spring-rabbit:$springAmqpVersion") {
             exclude group: 'org.springframework'
         }
+        optionalApi("org.springframework.amqp:spring-rabbit-stream:$springAmqpVersion") {
+            exclude group: 'org.springframework'
+        }
 
         testImplementation("org.springframework.amqp:spring-rabbit-junit:$springAmqpVersion") {
             exclude group: 'org.springframework'
         }
         testImplementation project(':spring-integration-stream')
         testImplementation 'org.springframework:spring-web'
+        testImplementation 'org.testcontainers:rabbitmq'
     }
 }
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,14 +28,14 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.ConditionalExceptionLogger;
 import org.springframework.amqp.support.ConsumerTagStrategy;
-import org.springframework.integration.dsl.IntegrationComponentSpec;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.interceptor.TransactionAttribute;
 import org.springframework.util.ErrorHandler;
 import org.springframework.util.backoff.BackOff;
 
 /**
- * Base class for container specs.
+ * Base class for container specs for containers that extend
+ * {@link AbstractMessageListenerContainer}.
  *
  * @param <S> the current spec extension type
  * @param <C> the listener container type
@@ -48,7 +48,7 @@ import org.springframework.util.backoff.BackOff;
  */
 public abstract class AbstractMessageListenerContainerSpec<S extends AbstractMessageListenerContainerSpec<S, C>,
 		C extends AbstractMessageListenerContainer>
-		extends IntegrationComponentSpec<S, C> {
+		extends MessageListenerContainerSpec<S, C> {
 
 	public AbstractMessageListenerContainerSpec(C listenerContainer) {
 		this.target = listenerContainer;

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,7 +19,7 @@ package org.springframework.integration.amqp.dsl;
 import java.util.Collections;
 import java.util.Map;
 
-import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.dsl.ComponentsRegistration;
 
@@ -36,13 +36,13 @@ import org.springframework.integration.dsl.ComponentsRegistration;
  * @since 5.0
  */
 public abstract class AmqpInboundChannelAdapterSpec
-			<S extends AmqpInboundChannelAdapterSpec<S, C>, C extends AbstractMessageListenerContainer>
+			<S extends AmqpInboundChannelAdapterSpec<S, C>, C extends MessageListenerContainer>
 		extends AmqpBaseInboundChannelAdapterSpec<S>
 		implements ComponentsRegistration {
 
-	protected final AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec; // NOSONAR final
+	protected final MessageListenerContainerSpec<?, C> listenerContainerSpec; // NOSONAR final
 
-	protected AmqpInboundChannelAdapterSpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec) {
+	protected AmqpInboundChannelAdapterSpec(MessageListenerContainerSpec<?, C> listenerContainerSpec) {
 		super(new AmqpInboundChannelAdapter(listenerContainerSpec.get()));
 		this.listenerContainerSpec = listenerContainerSpec;
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/MessageListenerContainerSpec.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.dsl;
+
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
+import org.springframework.integration.dsl.IntegrationComponentSpec;
+
+/**
+ * Base class for container specs.
+ *
+ * @param <S> the current spec extension type
+ * @param <C> the listener container type
+ *
+ * @author Gary Russell
+ *
+ * @since 6.0
+ *
+ */
+public abstract class MessageListenerContainerSpec<S extends MessageListenerContainerSpec<S, C>,
+		C extends MessageListenerContainer>
+		extends IntegrationComponentSpec<S, C> {
+
+	/**
+	 * Set the queue names.
+	 * @param queueNames the queue names.
+	 * @return this spec.
+	 */
+	public S queueName(String... queueNames) {
+		this.target.setQueueNames(queueNames);
+		return _this();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStream.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.dsl;
+
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
+
+import com.rabbitmq.stream.Codec;
+import com.rabbitmq.stream.Environment;
+
+/**
+ * Factory class for RabbitMQ components.
+ *
+ * @author Gary Russell
+ * @since 6.0
+ *
+ */
+public final class RabbitStream {
+
+	private RabbitStream() {
+	}
+
+	/**
+	 * Create an initial {@link RabbitStreamInboundChannelAdapterSpec}
+	 * with the provided {@link StreamListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link StreamListenerContainer} using
+	 * {@link RabbitStreamInboundChannelAdapterSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param listenerContainer the listenerContainer.
+	 * @return the RabbitInboundChannelAdapterSLCSpec.
+	 */
+	public static RabbitStreamInboundChannelAdapterSpec inboundAdapter(StreamListenerContainer listenerContainer) {
+		return new RabbitStreamInboundChannelAdapterSpec(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link RabbitStreamInboundChannelAdapterSpec}
+	 * with the provided {@link Environment}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link StreamListenerContainer} using
+	 * {@link RabbitStreamInboundChannelAdapterSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param environment the environment.
+	 * @return the RabbitInboundChannelAdapterSLCSpec.
+	 */
+	public static RabbitStreamInboundChannelAdapterSpec inboundAdapter(Environment environment) {
+		return new RabbitStreamInboundChannelAdapterSpec(environment, null);
+	}
+
+	/**
+	 * Create an initial {@link RabbitStreamInboundChannelAdapterSpec}
+	 * with the provided {@link Environment}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link StreamListenerContainer} using
+	 * {@link RabbitStreamInboundChannelAdapterSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param environment the environment.
+	 * @param codec the codec.
+	 * @return the RabbitInboundChannelAdapterSLCSpec.
+	 */
+	public static RabbitStreamInboundChannelAdapterSpec inboundAdapter(Environment environment, Codec codec) {
+		return new RabbitStreamInboundChannelAdapterSpec(environment, codec);
+	}
+
+	/**
+	 * Create an initial {@link RabbitStreamMessageHandlerSpec} (adapter).
+	 * @param template the amqpTemplate.
+	 * @return the RabbitStreamMessageHandlerSpec.
+	 */
+	public static RabbitStreamMessageHandlerSpec outboundStreamAdapter(RabbitStreamTemplate template) {
+		return new RabbitStreamMessageHandlerSpec(template);
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamInboundChannelAdapterSpec.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2017-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.dsl;
+
+import java.util.function.Consumer;
+
+import org.springframework.lang.Nullable;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+
+import com.rabbitmq.stream.Codec;
+import com.rabbitmq.stream.Environment;
+
+/**
+ * Spec for an inbound channel adapter with a {@link StreamListenerContainer}.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ */
+public class RabbitStreamInboundChannelAdapterSpec
+		extends AmqpInboundChannelAdapterSpec<RabbitStreamInboundChannelAdapterSpec, StreamListenerContainer> {
+
+	protected RabbitStreamInboundChannelAdapterSpec(StreamListenerContainer listenerContainer) {
+		super(new RabbitStreamMessageListenerContainerSpec(listenerContainer));
+	}
+
+	protected RabbitStreamInboundChannelAdapterSpec(Environment environment, @Nullable Codec codec) {
+		super(new RabbitStreamMessageListenerContainerSpec(environment, codec));
+	}
+
+	public RabbitStreamInboundChannelAdapterSpec configureContainer(
+			Consumer<RabbitStreamMessageListenerContainerSpec> configurer) {
+
+		configurer.accept((RabbitStreamMessageListenerContainerSpec) this.listenerContainerSpec);
+		return this;
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageHandlerSpec.java

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.dsl;
+
+import org.springframework.integration.amqp.outbound.RabbitStreamMessageHandler;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.dsl.MessageHandlerSpec;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.rabbit.stream.producer.RabbitStreamOperations;
+
+/**
+ * The base {@link MessageHandlerSpec} for {@link RabbitStreamMessageHandler}s.
+ *
+ * @author Gary Russell
+ *
+ * @since 6.0
+ */
+public class RabbitStreamMessageHandlerSpec
+		extends MessageHandlerSpec<RabbitStreamMessageHandlerSpec, RabbitStreamMessageHandler> {
+
+	private final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
+
+	RabbitStreamMessageHandlerSpec(RabbitStreamOperations operations) {
+		this.target = new RabbitStreamMessageHandler(operations);
+	}
+
+	/**
+	 * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers.
+	 * @param headerMapper the {@link AmqpHeaderMapper} to use.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageHandlerSpec headerMapper(AmqpHeaderMapper headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return this;
+	}
+
+	/**
+	 * Provide the header names that should be mapped from a request to a
+	 * {@link org.springframework.messaging.MessageHeaders}.
+	 * @param headers The request header names.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageHandlerSpec mappedRequestHeaders(String... headers) {
+		this.headerMapper.setRequestHeaderNames(headers);
+		return this;
+	}
+
+	/**
+	 * Determine whether the headers are
+	 * mapped before the message is converted, or afterwards.
+	 * @param headersLast true to map headers last.
+	 * @return this spec.
+	 * @see RabbitStreamMessageHandler#setHeadersMappedLast(boolean)
+	 */
+	public RabbitStreamMessageHandlerSpec headersMappedLast(boolean headersLast) {
+		this.target.setHeadersMappedLast(headersLast);
+		return this;
+	}
+
+	/**
+	 * Set the success channel.
+	 * @param channel the channel.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageHandlerSpec sendSuccessChannel(MessageChannel channel) {
+		this.target.setSendSuccessChannel(channel);
+		return this;
+	}
+
+	/**
+	 * Set the failure channel. After a send failure, an
+	 * {@link org.springframework.messaging.support.ErrorMessage} will be sent
+	 * to this channel with a payload of the exception with the
+	 * failed message.
+	 * @param channel the channel.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageHandlerSpec sendFailureChannel(MessageChannel channel) {
+		this.target.setSendFailureChannel(channel);
+		return this;
+	}
+
+	/**
+	 * Set the success channel.
+	 * @param channel the channel.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageHandlerSpec sendSuccessChannel(String channel) {
+		this.target.setSendSuccessChannelName(channel);
+		return this;
+	}
+
+	/**
+	 * Set the failure channel. After a send failure, an
+	 * {@link org.springframework.messaging.support.ErrorMessage} will be sent
+	 * to this channel with a payload of the exception with the
+	 * failed message.
+	 * @param channel the channel.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageHandlerSpec sendFailureChannel(String channel) {
+		this.target.setSendFailureChannelName(channel);
+		return this;
+	}
+
+	/**
+	 * Set to true to wait for a confirmation.
+	 * @param sync true to wait.
+	 * @return this spec.
+	 * @see #setConfirmTimeout(long)
+	 */
+	public RabbitStreamMessageHandlerSpec sync(boolean sync) {
+		this.target.setSync(sync);
+		return this;
+	}
+
+	/**
+	 * Set a timeout for the confirm result.
+	 * @param timeout the approximate timeout.
+	 * @return this spec.
+	 * @see #sync(boolean)
+	 */
+	public RabbitStreamMessageHandlerSpec confirmTimeout(long timeout) {
+		this.target.setConfirmTimeout(timeout);
+		return this;
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageListenerContainerSpec.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.dsl;
+
+import java.util.function.Consumer;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.lang.Nullable;
+import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
+
+import com.rabbitmq.stream.Codec;
+import com.rabbitmq.stream.Environment;
+
+/**
+ * Spec for {@link StreamListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 6.0
+ *
+ */
+public class RabbitStreamMessageListenerContainerSpec extends
+		MessageListenerContainerSpec<RabbitStreamMessageListenerContainerSpec, StreamListenerContainer> {
+
+	RabbitStreamMessageListenerContainerSpec(StreamListenerContainer container) {
+		this.target = container;
+	}
+
+	RabbitStreamMessageListenerContainerSpec(Environment environment, @Nullable Codec codec) {
+		this.target = new StreamListenerContainer(environment, codec);
+	}
+
+	/**
+	 * Set the Stream queue name;
+	 * Mutually exclusive with {@link #superStream(String, String)}.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageListenerContainerSpec queueName(String queueName) {
+		return super.queueName(queueName);
+	}
+
+	/**
+	 * Enable Single Active Consumer on a Super Stream.
+	 * Mutually exclusive with {@link #setQueueName(String...)}.
+	 * @param superStream the stream.
+	 * @param name the consumer name.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageListenerContainerSpec superStream(String superStream, String name) {
+		this.target.superStream(superStream, name);
+		return this;
+	}
+
+	/**
+	 * Set a stream message converter.
+	 * @param converter the converter.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageListenerContainerSpec streamConverter(StreamMessageConverter converter) {
+		this.target.setStreamConverter(converter);
+		return this;
+	}
+
+	/**
+	 * Set a consumer customizer.
+	 * @param customizer the customizer.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageListenerContainerSpec consumerCustomizer(ConsumerCustomizer customizer) {
+		this.target.setConsumerCustomizer(customizer);
+		return this;
+	}
+
+	/**
+	 * @param adviceChain the adviceChain.
+	 * @return the spec.
+	 * @see StreamListenerContainer#setAdviceChain(Advice[])
+	 */
+	public RabbitStreamMessageListenerContainerSpec adviceChain(Advice... adviceChain) {
+		this.target.setAdviceChain(adviceChain);
+		return this;
+	}
+
+	/**
+	 * Perform additional configuration of the container.
+	 * @param consumer a consumer for the container.
+	 * @return this spec.
+	 */
+	public RabbitStreamMessageListenerContainerSpec configure(Consumer<StreamListenerContainer> consumer) {
+		consumer.accept(this.target);
+		return this;
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandler.java

@@ -0,0 +1,249 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.outbound;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.amqp.support.MappingUtils;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.rabbit.stream.producer.RabbitStreamOperations;
+import org.springframework.rabbit.stream.support.StreamMessageProperties;
+import org.springframework.util.Assert;
+
+/**
+ * {@link MessageHandler} based on {@link RabbitStreamOperations}.
+ *
+ * @author Gary Russell
+ * @author Chris Bono
+ * @since 6.0
+ *
+ */
+public class RabbitStreamMessageHandler extends AbstractMessageHandler {
+
+	private static final int DEFAULT_CONFIRM_TIMEOUT = 10_000;
+
+	private final RabbitStreamOperations streamOperations;
+
+	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+	private boolean sync;
+
+	private long confirmTimeout = DEFAULT_CONFIRM_TIMEOUT;
+
+	private MessageChannel sendFailureChannel;
+
+	private String sendFailureChannelName;
+
+	private MessageChannel sendSuccessChannel;
+
+	private String sendSuccessChannelName;
+
+	private AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
+
+	private boolean headersMappedLast;
+
+	/**
+	 * Create an instance with the provided {@link RabbitStreamOperations}.
+	 * @param streamOperations the operations.
+	 */
+	public RabbitStreamMessageHandler(RabbitStreamOperations streamOperations) {
+		Assert.notNull(streamOperations, "'streamOperations' cannot be null");
+		this.streamOperations = streamOperations;
+	}
+
+	/**
+	 * Set the failure channel. After a send failure, an
+	 * {@link org.springframework.messaging.support.ErrorMessage} will be sent
+	 * to this channel with a payload of the exception with the
+	 * failed message.
+	 * @param sendFailureChannel the failure channel.
+	 */
+	public void setSendFailureChannel(MessageChannel sendFailureChannel) {
+		this.sendFailureChannel = sendFailureChannel;
+	}
+
+	/**
+	 * Set the failure channel name. After a send failure, an
+	 * {@link org.springframework.messaging.support.ErrorMessage} will be sent
+	 * to this channel with a payload of the exception with the
+	 * failed message.
+	 * @param sendFailureChannelName the failure channel name.
+	 */
+	public void setSendFailureChannelName(String sendFailureChannelName) {
+		this.sendFailureChannelName = sendFailureChannelName;
+	}
+
+	/**
+	 * Set the success channel.
+	 * @param sendSuccessChannel the success channel.
+	 */
+	public void setSendSuccessChannel(MessageChannel sendSuccessChannel) {
+		this.sendSuccessChannel = sendSuccessChannel;
+	}
+
+	/**
+	 * Set the Success channel name.
+	 * @param sendSuccessChannelName the success channel name.
+	 */
+	public void setSendSuccessChannelName(String sendSuccessChannelName) {
+		this.sendSuccessChannelName = sendSuccessChannelName;
+	}
+
+	/**
+	 * Set to true to wait for a confirmation.
+	 * @param sync true to wait.
+	 * @see #setConfirmTimeout(long)
+	 */
+	public void setSync(boolean sync) {
+		this.sync = sync;
+	}
+
+	/**
+	 * Set the confirm timeout.
+	 * @param confirmTimeout the timeout.
+	 * @see #setSync(boolean)
+	 */
+	public void setConfirmTimeout(long confirmTimeout) {
+		this.confirmTimeout = confirmTimeout;
+	}
+
+	/**
+	 * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers.
+	 * Defaults to {@link DefaultAmqpHeaderMapper#outboundMapper()}.
+	 * @param headerMapper the {@link AmqpHeaderMapper} to use.
+	 */
+	public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
+		Assert.notNull(headerMapper, "headerMapper must not be null");
+		this.headerMapper = headerMapper;
+	}
+
+	/**
+	 * When mapping headers for the outbound message, determine whether the headers are
+	 * mapped before the message is converted, or afterwards. This only affects headers
+	 * that might be added by the message converter. When false, the converter's headers
+	 * win; when true, any headers added by the converter will be overridden (if the
+	 * source message has a header that maps to those headers). You might wish to set this
+	 * to true, for example, when using a
+	 * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a
+	 * String payload that contains json; the converter will set the content type to
+	 * {@code text/plain} which can be overridden to {@code application/json} by setting
+	 * the {@link AmqpHeaders#CONTENT_TYPE} message header. Default: false.
+	 * @param headersMappedLast true if headers are mapped after conversion.
+	 */
+	public void setHeadersMappedLast(boolean headersMappedLast) {
+		this.headersMappedLast = headersMappedLast;
+	}
+
+	/**
+	 * Return the {@link RabbitStreamOperations}.
+	 * @return the operations.
+	 */
+	public RabbitStreamOperations getStreamOperations() {
+		return this.streamOperations;
+	}
+
+	protected MessageChannel getSendFailureChannel() {
+		if (this.sendFailureChannel != null) {
+			return this.sendFailureChannel;
+		}
+		else if (this.sendFailureChannelName != null) {
+			this.sendFailureChannel = getChannelResolver().resolveDestination(this.sendFailureChannelName);
+			return this.sendFailureChannel;
+		}
+		return null;
+	}
+
+	protected MessageChannel getSendSuccessChannel() {
+		if (this.sendSuccessChannel != null) {
+			return this.sendSuccessChannel;
+		}
+		else if (this.sendSuccessChannelName != null) {
+			this.sendSuccessChannel = getChannelResolver().resolveDestination(this.sendSuccessChannelName);
+			return this.sendSuccessChannel;
+		}
+		return null;
+	}
+
+	@Override
+	protected void handleMessageInternal(Message<?> requestMessage) {
+		CompletableFuture<Boolean> future;
+		com.rabbitmq.stream.Message streamMessage;
+		if (requestMessage.getPayload() instanceof com.rabbitmq.stream.Message) {
+			streamMessage = (com.rabbitmq.stream.Message) requestMessage.getPayload();
+		}
+		else {
+			MessageConverter converter = this.streamOperations.messageConverter();
+			org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter,
+					this.headerMapper, this.headersMappedLast);
+			streamMessage = this.streamOperations.streamMessageConverter().fromMessage(amqpMessage);
+		}
+		future = this.streamOperations.send(streamMessage);
+		handleConfirms(requestMessage, future);
+	}
+
+	private void handleConfirms(Message<?> message, CompletableFuture<Boolean> future) {
+		future.whenComplete((bool, ex) -> {
+			if (ex != null) {
+				MessageChannel failures = getSendFailureChannel();
+				if (failures != null) {
+					this.messagingTemplate.send(failures, new ErrorMessage(ex, message));
+				}
+			}
+			else {
+				MessageChannel successes = getSendSuccessChannel();
+				if (successes != null) {
+					this.messagingTemplate.send(successes, message);
+				}
+			}
+		});
+		if (this.sync) {
+			try {
+				future.get(this.confirmTimeout, TimeUnit.MILLISECONDS);
+			}
+			catch (InterruptedException ex) {
+				Thread.currentThread().interrupt();
+				throw new MessageHandlingException(message, ex);
+			}
+			catch (ExecutionException | TimeoutException ex) {
+				throw new MessageHandlingException(message, ex);
+			}
+		}
+	}
+
+	private static org.springframework.amqp.core.Message mapMessage(Message<?> message,
+			MessageConverter converter, AmqpHeaderMapper headerMapper, boolean headersMappedLast) {
+
+		MessageProperties amqpMessageProperties = new StreamMessageProperties();
+		return MappingUtils.mapMessage(message, converter, headerMapper, headersMappedLast, headersMappedLast,
+				amqpMessageProperties);
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -73,17 +73,39 @@ public final class MappingUtils {
 	 * @since 5.1.9
 	 */
 	public static org.springframework.amqp.core.Message mapReplyMessage(Message<?> replyMessage,
-			MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode,
-			boolean headersMappedLast) {
+			MessageConverter converter, AmqpHeaderMapper headerMapper,
+			@Nullable MessageDeliveryMode defaultDeliveryMode, boolean headersMappedLast) {
 
 		return doMapMessage(replyMessage, converter, headerMapper, defaultDeliveryMode, headersMappedLast, true);
 	}
 
 	private static org.springframework.amqp.core.Message doMapMessage(Message<?> message,
-			MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode,
-			boolean headersMappedLast, boolean reply) {
+			MessageConverter converter, AmqpHeaderMapper headerMapper,
+			@Nullable MessageDeliveryMode defaultDeliveryMode, boolean headersMappedLast, boolean reply) {
 
 		MessageProperties amqpMessageProperties = new MessageProperties();
+		org.springframework.amqp.core.Message amqpMessage = mapMessage(message, converter, headerMapper,
+				headersMappedLast, reply, amqpMessageProperties);
+		checkDeliveryMode(message, amqpMessageProperties, defaultDeliveryMode);
+		return amqpMessage;
+	}
+
+	/**
+	 * Map a reply o.s.m.Message to an o.s.a.core.Message. When using a
+	 * {@link ContentTypeDelegatingMessageConverter}, {@link AmqpHeaders#CONTENT_TYPE} and
+	 * {@link MessageHeaders#CONTENT_TYPE} will be used for the selection, with the AMQP
+	 * header taking precedence.
+	 * @param replyMessage the reply message.
+	 * @param converter the message converter to use.
+	 * @param headerMapper the header mapper to use.
+	 * @param headersMappedLast true if headers are mapped after conversion.
+	 * @return the mapped Message.
+	 * @since 6.0
+	 */
+	public static org.springframework.amqp.core.Message mapMessage(Message<?> message, MessageConverter converter,
+			AmqpHeaderMapper headerMapper, boolean headersMappedLast, boolean reply,
+			MessageProperties amqpMessageProperties) {
+
 		org.springframework.amqp.core.Message amqpMessage;
 		if (!headersMappedLast) {
 			mapHeaders(message.getHeaders(), amqpMessageProperties, headerMapper, reply);
@@ -98,7 +120,6 @@ public final class MappingUtils {
 		if (headersMappedLast) {
 			mapHeaders(message.getHeaders(), amqpMessageProperties, headerMapper, reply);
 		}
-		checkDeliveryMode(message, amqpMessageProperties, defaultDeliveryMode);
 		return amqpMessage;
 	}
 

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java

@@ -17,6 +17,10 @@
 package org.springframework.integration.amqp.dsl;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -67,9 +71,13 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
+import com.rabbitmq.stream.ConsumerBuilder;
+import com.rabbitmq.stream.Environment;
+
 /**
  * @author Artem Bilan
  * @author Gary Russell
@@ -263,6 +271,17 @@ public class AmqpTests {
 		registration.destroy();
 	}
 
+	@Test
+	void streamContainer() {
+		Environment env = mock(Environment.class);
+		given(env.consumerBuilder()).willReturn(mock(ConsumerBuilder.class));
+		RabbitStreamInboundChannelAdapterSpec inboundAdapter = RabbitStream.inboundAdapter(env);
+		ConsumerCustomizer customizer = mock(ConsumerCustomizer.class);
+		inboundAdapter.configureContainer(container -> container.consumerCustomizer(customizer));
+		inboundAdapter.start();
+		verify(customizer).accept(any(), any());
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandlerTests.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2021-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.outbound;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.amqp.dsl.RabbitStream;
+import org.springframework.integration.amqp.support.RabbitTestContainer;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
+
+import com.rabbitmq.stream.Address;
+import com.rabbitmq.stream.Consumer;
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.OffsetSpecification;
+
+/**
+ * @author Gary Russell
+ * @author Chris Bono
+ * @since 6.0
+ */
+public class RabbitStreamMessageHandlerTests implements RabbitTestContainer {
+
+	@Test
+	void convertAndSend() throws InterruptedException {
+		Environment env = Environment.builder()
+				.lazyInitialization(true)
+				.addressResolver(add -> new Address("localhost", RabbitTestContainer.streamPort()))
+				.build();
+		try {
+			env.deleteStream("stream.stream");
+		}
+		catch (Exception e) {
+		}
+		env.streamCreator().stream("stream.stream").create();
+		RabbitStreamTemplate streamTemplate = new RabbitStreamTemplate(env, "stream.stream");
+
+		RabbitStreamMessageHandler handler = RabbitStream.outboundStreamAdapter(streamTemplate)
+				.sync(true)
+				.get();
+
+		handler.handleMessage(MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.build());
+		CountDownLatch latch = new CountDownLatch(1);
+		AtomicReference<com.rabbitmq.stream.Message> received = new AtomicReference<>();
+		Consumer consumer = env.consumerBuilder().stream("stream.stream")
+				.offset(OffsetSpecification.first())
+				.messageHandler((context, msg) -> {
+					received.set(msg);
+					latch.countDown();
+				})
+				.build();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(received.get()).isNotNull();
+		assertThat(received.get().getBodyAsBinary()).isEqualTo("foo".getBytes());
+		assertThat((String) received.get().getApplicationProperties().get("bar")).isEqualTo("baz");
+		consumer.close();
+		streamTemplate.close();
+	}
+
+	@Test
+	void sendNative() throws InterruptedException {
+		Environment env = Environment.builder()
+				.addressResolver(add -> new Address("localhost", RabbitTestContainer.streamPort()))
+				.lazyInitialization(true)
+				.build();
+		try {
+			env.deleteStream("stream.stream");
+		}
+		catch (Exception e) {
+		}
+		env.streamCreator().stream("stream.stream").create();
+		RabbitStreamTemplate streamTemplate = new RabbitStreamTemplate(env, "stream.stream");
+		RabbitStreamMessageHandler handler = new RabbitStreamMessageHandler(streamTemplate);
+		handler.setSync(true);
+		handler.handleMessage(MessageBuilder.withPayload(streamTemplate.messageBuilder()
+						.addData("foo".getBytes())
+						.applicationProperties().entry("bar", "baz")
+						.messageBuilder()
+						.build())
+				.build());
+		CountDownLatch latch = new CountDownLatch(1);
+		AtomicReference<com.rabbitmq.stream.Message> received = new AtomicReference<>();
+		Consumer consumer = env.consumerBuilder().stream("stream.stream")
+				.offset(OffsetSpecification.first())
+				.messageHandler((context, msg) -> {
+					received.set(msg);
+					latch.countDown();
+				})
+				.build();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(received.get()).isNotNull();
+		assertThat(received.get().getBodyAsBinary()).isEqualTo("foo".getBytes());
+		assertThat((String) received.get().getApplicationProperties().get("bar")).isEqualTo("baz");
+		consumer.close();
+		streamTemplate.close();
+	}
+
+}

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/RabbitTestContainer.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.support;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.RabbitMQContainer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+/**
+ * Provides a static {@link RabbitMQContainer} that can be shared across test classes.
+ *
+ * @author Chris Bono
+ * @author Gary Russell
+ */
+@Testcontainers(disabledWithoutDocker = true)
+public interface RabbitTestContainer {
+
+	RabbitMQContainer RABBITMQ = new RabbitMQContainer("rabbitmq:management")
+			.withExposedPorts(5672, 15672, 5552)
+			.withEnv("RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS", "-rabbitmq_stream advertised_host localhost")
+			.withPluginsEnabled("rabbitmq_stream")
+			.withStartupTimeout(Duration.ofMinutes(2));
+
+	@BeforeAll
+	static void startContainer() {
+		RABBITMQ.start();
+	}
+
+	static int amqpPort() {
+		return RABBITMQ.getMappedPort(5672);
+	}
+
+	static int managementPort() {
+		return RABBITMQ.getMappedPort(15672);
+	}
+
+	static int streamPort() {
+		return RABBITMQ.getMappedPort(5552);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java

@@ -162,7 +162,6 @@ public class IntegrationManagementConfigurer
 		return this.metricsCaptor;
 	}
 
-	@Nullable
 	private void setupObservationRegistry() {
 		if (this.observationRegistry == null && this.observationRegistryProvider != null) {
 			this.observationRegistry = this.observationRegistryProvider.getIfUnique();

src/reference/asciidoc/amqp.adoc

@@ -1,5 +1,5 @@
 [[amqp]]
-== AMQP Support
+== AMQP (RabbitMQ) Support
 
 Spring Integration provides channel adapters for receiving and sending messages by using the Advanced Message Queuing Protocol (AMQP).
 
@@ -29,6 +29,8 @@ The following adapters are available:
 * <<amqp-outbound-channel-adapter,Outbound Channel Adapter>>
 * <<amqp-outbound-gateway,Outbound Gateway>>
 * <<amqp-async-outbound-gateway,Async Outbound Gateway>>
+* <<rmq-stream-inbound-channel-adapter,RabbitMQ Stream Queue Inbound Channel Adapter>>
+* <<rmq-stream-outbound-channel-adapter,RabbitMQ Stream Queue Outbound Channel Adapter>>
 
 Spring Integration also provides a point-to-point message channel and a publish-subscribe message channel backed by AMQP Exchanges and Queues.
 
@@ -1417,3 +1419,53 @@ In return, that message is retrieved by Spring Integration and printed to the co
 The following image illustrates the basic set of Spring Integration components used in this sample.
 
 .The Spring Integration graph of the AMQP sample image::images/spring-integration-amqp-sample-graph.png[]
+
+[[rmq-streams]]
+=== RabbitMQ Stream Queue Support
+
+Version 6.0 introduced support for RabbitMQ Stream Queues.
+
+The DSL factory class for these endpoints is `Rabbit`.
+
+[[rmq-stream-inbound-channel-adapter]]
+==== RabbitMQ Stream Inbound Channel Adapter
+
+====
+[source, java]
+----
+@Bean
+IntegrationFlow flow(Environment env) {
+    @Bean
+	IntegrationFlow simpleStream(Environment env) {
+		return IntegrationFlow.from(RabbitStream.inboundAdapter(env)
+						.configureContainer(container -> container.queueName("my.stream")))
+				// ...
+				.get();
+	}
+
+	@Bean
+	IntegrationFlow superStream(Environment env) {
+		return IntegrationFlow.from(RabbitStream.inboundAdapter(env)
+						.configureContainer(container -> container.superStream("my.stream", "my.consumer")))
+				// ...
+				.get();
+	}
+}
+----
+====
+
+[[rmq-stream-outbound-channel-adapter]]
+==== RabbitMQ Stream Outbound Channel Adapter
+
+====
+[source, java]
+----
+@Bean
+IntegrationFlow outbound(RabbitStreamTemplate template) {
+    return f -> f
+            // ...
+            .handle(RabbitStream.outboundStreamAdapter(template));
+
+}
+----
+====

src/reference/asciidoc/whats-new.adoc

@@ -44,6 +44,11 @@ A `PostgresSubscribableChannel` allows to receive push notifications via `Postgr
 
 See <<./jdbc.adoc#postgresql-push,PostgreSQL: Receiving Push Notifications>> for more information.
 
+[[x6.0-rmq]]
+==== RabbitMQ Stream Queue Support
+
+The AMQP module has been enhanced to provide support for inbound and outbound channel adapters using RabbitMQ Stream Queues.
+See <<./amqp.adoc#rmq-streams,RabbitMQ Stream Queue Support>> for more information.
 
 [[x6.0-general]]
 === General Changes