GH-3626: RabbitMQ Stream Support

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

@@ -25,11 +25,6 @@
 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.
@@ -244,49 +239,6 @@ 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.
@@ -296,15 +248,6 @@ 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/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/RabbitInboundChannelAdapterSLCSpec.java → spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamInboundChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 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.
@@ -18,34 +18,33 @@
 
 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}.
+ * Spec for an inbound channel adapter with a {@link StreamListenerContainer}.
  *
  * @author Gary Russell
  * @author Artem Bilan
  *
  * @since 6.0
  *
  */
-public class RabbitInboundChannelAdapterSLCSpec
-		extends AmqpInboundChannelAdapterSpec<RabbitInboundChannelAdapterSLCSpec, StreamListenerContainer> {
+public class RabbitStreamInboundChannelAdapterSpec
+		extends AmqpInboundChannelAdapterSpec<RabbitStreamInboundChannelAdapterSpec, StreamListenerContainer> {
 
-	protected RabbitInboundChannelAdapterSLCSpec(StreamListenerContainer listenerContainer) {
+	protected RabbitStreamInboundChannelAdapterSpec(StreamListenerContainer listenerContainer) {
 		super(new RabbitStreamMessageListenerContainerSpec(listenerContainer));
 	}
 
-	protected RabbitInboundChannelAdapterSLCSpec(Environment environment, @Nullable Codec codec) {
+	protected RabbitStreamInboundChannelAdapterSpec(Environment environment, @Nullable Codec codec) {
 		super(new RabbitStreamMessageListenerContainerSpec(environment, codec));
 	}
 
-	public RabbitInboundChannelAdapterSLCSpec configureContainer(
+	public RabbitStreamInboundChannelAdapterSpec configureContainer(
 			Consumer<RabbitStreamMessageListenerContainerSpec> configurer) {
 
 		configurer.accept((RabbitStreamMessageListenerContainerSpec) this.listenerContainerSpec);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * 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.
@@ -16,15 +16,15 @@
 
 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.messaging.MessageChannel;
 import org.springframework.rabbit.stream.producer.RabbitStreamOperations;
 
 /**
- * The base {@link MessageHandlerSpec} for {@link AbstractAmqpOutboundEndpoint}s.
+ * The base {@link MessageHandlerSpec} for {@link RabbitStreamMessageHandler}s.
  *
  * @author Gary Russell
  *
@@ -42,7 +42,7 @@ public class RabbitStreamMessageHandlerSpec
 	/**
 	 * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers.
 	 * @param headerMapper the {@link AmqpHeaderMapper} to use.
-	 * @return the spec
+	 * @return this spec.
 	 */
 	public RabbitStreamMessageHandlerSpec headerMapper(AmqpHeaderMapper headerMapper) {
 		this.target.setHeaderMapper(headerMapper);
@@ -53,7 +53,7 @@ public RabbitStreamMessageHandlerSpec headerMapper(AmqpHeaderMapper headerMapper
 	 * 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
+	 * @return this spec.
 	 */
 	public RabbitStreamMessageHandlerSpec mappedRequestHeaders(String... headers) {
 		this.headerMapper.setRequestHeaderNames(headers);
@@ -64,35 +64,64 @@ public RabbitStreamMessageHandlerSpec mappedRequestHeaders(String... headers) {
 	 * 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)
+	 * @return this spec.
+	 * @see RabbitStreamMessageHandler#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.
+	 * Set the success channel.
+	 * @param channel the channel.
+	 * @return this spec.
 	 */
-	public RabbitStreamMessageHandlerSpec successCallback(RabbitStreamMessageHandler.SuccessCallback callback) {
-		this.target.setSuccessCallback(callback);
+	public RabbitStreamMessageHandlerSpec sendSuccessChannel(MessageChannel channel) {
+		this.target.setSendSuccessChannel(channel);
 		return this;
 	}
 
 	/**
-	 * Set a callback to be invoked when a send fails.
-	 * @param callback the callback.
+	 * 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 failureCallback(RabbitStreamMessageHandler.FailureCallback callback) {
-		this.target.setFailureCallback(callback);
+	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) {
@@ -103,7 +132,7 @@ public RabbitStreamMessageHandlerSpec sync(boolean sync) {
 	/**
 	 * Set a timeout for the confirm result.
 	 * @param timeout the approximate timeout.
-	 * @return the spec.
+	 * @return this spec.
 	 * @see #sync(boolean)
 	 */
 	public RabbitStreamMessageHandlerSpec confirmTimeout(long timeout) {