GH-3626: RabbitMQ Stream Support

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.

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/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;
+	}
+
+}