spring-atticGH-325: Initial Support for RabbitMQ Stream Plugin

Resolves spring-attic#325

Author: garyrussell

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,39 @@ 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.
+
+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.
+
+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());
+        });
+        // ...
+    };
+}
+----
+====
+
+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/pom.xml

@@ -25,6 +25,7 @@
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-amqp</artifactId>
+			<version>5.5.2</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

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

@@ -20,7 +20,6 @@
 
 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 void setReceiveTimeout(Long receiveTimeout) {
 		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.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 @@ private ConsumerDestination doProvisionConsumerDestination(String name, String g
 		}
 		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

@@ -93,7 +93,7 @@ public int getPort() {
 
 		public void start() throws IOException {
 			this.serverSocket = ServerSocketFactory.getDefault()
-					.createServerSocket(this.port);
+					.createServerSocket(this.port, 10);
 			this.serverExec.execute(new Runnable() {
 
 				@Override

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

@@ -55,13 +55,41 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-amqp</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.amqp</groupId>
+			<artifactId>spring-amqp</artifactId>
+			<version>2.4.0-M1</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.amqp</groupId>
+			<artifactId>spring-rabbit</artifactId>
+			<version>2.4.0-M1</version>
+		</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>
+			<version>5.5.2</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-amqp</artifactId>
+			<version>5.5.2</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+			<version>5.5.2</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-test</artifactId>
+			<version>5.5.2</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>

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.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.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.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,15 +187,15 @@ public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
 			RabbitProperties rabbitProperties,
 			RabbitExchangeQueueProvisioner provisioningProvider,
-			ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer) {
+			ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer) {
 
 		this(connectionFactory, rabbitProperties, provisioningProvider, containerCustomizer, null);
 	}
 
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
 			RabbitProperties rabbitProperties,
 			RabbitExchangeQueueProvisioner provisioningProvider,
-			ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer,
+			ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer,
 			MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer) {
 
 		super(new String[0], provisioningProvider, containerCustomizer, sourceCustomizer);
@@ -460,6 +461,47 @@ protected MessageProducer createConsumerEndpoint(
 				"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);
+		}
+		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 +527,6 @@ protected MessageProducer createConsumerEndpoint(
 				.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 +544,13 @@ protected MessageProducer createConsumerEndpoint(
 		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(