GH-3626: RabbitMQ Stream Support

Resolves #3626

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

Author: garyrussell

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.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 @@
  */
 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.
@@ -25,6 +25,11 @@
 import org.springframework.integration.amqp.channel.PollableAmqpChannel;
 import org.springframework.integration.amqp.inbound.AmqpMessageSource.AmqpAckCallbackFactory;
 import org.springframework.lang.Nullable;
+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 AMQP components.
@@ -239,6 +244,49 @@ public static AmqpInboundChannelAdapterDMLCSpec inboundAdapter(DirectMessageList
 		return new AmqpInboundChannelAdapterDMLCSpec(listenerContainer);
 	}
 
+	/**
+	 * Create an initial {@link RabbitInboundChannelAdapterSLCSpec}
+	 * 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 RabbitInboundChannelAdapterSLCSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param listenerContainer the listenerContainer.
+	 * @return the RabbitInboundChannelAdapterSLCSpec.
+	 */
+	public static RabbitInboundChannelAdapterSLCSpec inboundAdapter(StreamListenerContainer listenerContainer) {
+		return new RabbitInboundChannelAdapterSLCSpec(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link RabbitInboundChannelAdapterSLCSpec}
+	 * 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 RabbitInboundChannelAdapterSLCSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param environment the environment.
+	 * @return the RabbitInboundChannelAdapterSLCSpec.
+	 */
+	public static RabbitInboundChannelAdapterSLCSpec inboundAdapter(Environment environment) {
+		return new RabbitInboundChannelAdapterSLCSpec(environment, null);
+	}
+
+	/**
+	 * Create an initial {@link RabbitInboundChannelAdapterSLCSpec}
+	 * 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 RabbitInboundChannelAdapterSLCSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param environment the environment.
+	 * @param codec the codec.
+	 * @return the RabbitInboundChannelAdapterSLCSpec.
+	 */
+	public static RabbitInboundChannelAdapterSLCSpec inboundAdapter(Environment environment, Codec codec) {
+		return new RabbitInboundChannelAdapterSLCSpec(environment, codec);
+	}
+
 	/**
 	 * Create an initial AmqpOutboundEndpointSpec (adapter).
 	 * @param amqpTemplate the amqpTemplate.
@@ -248,6 +296,15 @@ public static AmqpOutboundChannelAdapterSpec outboundAdapter(AmqpTemplate amqpTe
 		return new AmqpOutboundChannelAdapterSpec(amqpTemplate);
 	}
 
+	/**
+	 * Create an initial {@link RabbitStreamMessageHandlerSpec} (adapter).
+	 * @param template the amqpTemplate.
+	 * @return the RabbitStreamMessageHandlerSpec.
+	 */
+	public static RabbitStreamMessageHandlerSpec outboundStreamAdapter(RabbitStreamTemplate template) {
+		return new RabbitStreamMessageHandlerSpec(template);
+	}
+
 	/**
 	 * Create an initial AmqpOutboundEndpointSpec (gateway).
 	 * @param amqpTemplate the amqpTemplate.

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 @@
 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 @@
  * @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/RabbitInboundChannelAdapterSLCSpec.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2017-2020 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.amqp.rabbit.listener.DirectMessageListenerContainer;
+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 DirectMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ */
+public class RabbitInboundChannelAdapterSLCSpec
+		extends AmqpInboundChannelAdapterSpec<RabbitInboundChannelAdapterSLCSpec, StreamListenerContainer> {
+
+	protected RabbitInboundChannelAdapterSLCSpec(StreamListenerContainer listenerContainer) {
+		super(new RabbitStreamMessageListenerContainerSpec(listenerContainer));
+	}
+
+	protected RabbitInboundChannelAdapterSLCSpec(Environment environment, @Nullable Codec codec) {
+		super(new RabbitStreamMessageListenerContainerSpec(environment, codec));
+	}
+
+	public RabbitInboundChannelAdapterSLCSpec 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,114 @@
+/*
+ * Copyright 2016-2020 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.AbstractAmqpOutboundEndpoint;
+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.rabbit.stream.producer.RabbitStreamOperations;
+
+/**
+ * The base {@link MessageHandlerSpec} for {@link AbstractAmqpOutboundEndpoint}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 the 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 the 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 the spec.
+	 * @see AbstractAmqpOutboundEndpoint#setHeadersMappedLast(boolean)
+	 */
+	public RabbitStreamMessageHandlerSpec headersMappedLast(boolean headersLast) {
+		this.target.setHeadersMappedLast(headersLast);
+		return this;
+	}
+
+	/**
+	 * Set a callback to be invoked when a send is successful.
+	 * @param callback the callback.
+	 */
+	public RabbitStreamMessageHandlerSpec successCallback(RabbitStreamMessageHandler.SuccessCallback callback) {
+		this.target.setSuccessCallback(callback);
+		return this;
+	}
+
+	/**
+	 * Set a callback to be invoked when a send fails.
+	 * @param callback the callback.
+	 */
+	public RabbitStreamMessageHandlerSpec failureCallback(RabbitStreamMessageHandler.FailureCallback callback) {
+		this.target.setFailureCallback(callback);
+		return this;
+	}
+
+	/**
+	 * Set to true to wait for a confirmation.
+	 * @param sync true to wait.
+	 * @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 the spec.
+	 * @see #sync(boolean)
+	 */
+	public RabbitStreamMessageHandlerSpec confirmTimeout(long timeout) {
+		this.target.setConfirmTimeout(timeout);
+		return this;
+	}
+
+}