Add auto-configuration for spring-rabbit-stream

See gh-27480

Author: garyrussell

spring-boot-project/spring-boot-autoconfigure/build.gradle

@@ -142,6 +142,7 @@ dependencies {
 	optional("org.springframework.session:spring-session-hazelcast")
 	optional("org.springframework.session:spring-session-jdbc")
 	optional("org.springframework.amqp:spring-rabbit")
+	optional("org.springframework.amqp:spring-rabbit-stream")
 	optional("org.springframework.kafka:spring-kafka")
 	optional("org.springframework.ws:spring-ws-core")
 	optional("org.thymeleaf:thymeleaf")

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration.java

@@ -88,7 +88,7 @@
 @Configuration(proxyBeanMethods = false)
 @ConditionalOnClass({ RabbitTemplate.class, Channel.class })
 @EnableConfigurationProperties(RabbitProperties.class)
-@Import(RabbitAnnotationDrivenConfiguration.class)
+@Import({ RabbitAnnotationDrivenConfiguration.class, RabbitStreamConfiguration.class })
 public class RabbitAutoConfiguration {
 
 	@Configuration(proxyBeanMethods = false)

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java

@@ -51,6 +51,8 @@ public class RabbitProperties {
 
 	private static final int DEFAULT_PORT_SECURE = 5671;
 
+	private static final int DEFAULT_STREAM_PORT = 5552;
+
 	/**
 	 * RabbitMQ host. Ignored if an address is set.
 	 */
@@ -137,6 +139,8 @@ public class RabbitProperties {
 
 	private final Template template = new Template();
 
+	private final Stream stream = new Stream();
+
 	private List<Address> parsedAddresses;
 
 	public String getHost() {
@@ -361,6 +365,10 @@ public Template getTemplate() {
 		return this.template;
 	}
 
+	public Stream getStream() {
+		return this.stream;
+	}
+
 	public class Ssl {
 
 		private static final String SUN_X509 = "SunX509";
@@ -629,7 +637,12 @@ public enum ContainerType {
 		 * Container where the listener is invoked directly on the RabbitMQ consumer
 		 * thread.
 		 */
-		DIRECT
+		DIRECT,
+
+		/**
+		 * Container that uses the RabbitMQ Stream Client.
+		 */
+		STREAM
 
 	}
 
@@ -644,6 +657,8 @@ public static class Listener {
 
 		private final DirectContainer direct = new DirectContainer();
 
+		private final StreamContainer stream = new StreamContainer();
+
 		public ContainerType getType() {
 			return this.type;
 		}
@@ -660,15 +675,31 @@ public DirectContainer getDirect() {
 			return this.direct;
 		}
 
+		public StreamContainer getStream() {
+			return this.stream;
+		}
+
 	}
 
-	public abstract static class AmqpContainer {
+	public abstract static class BaseContainer {
 
 		/**
 		 * Whether to start the container automatically on startup.
 		 */
 		private boolean autoStartup = true;
 
+		public boolean isAutoStartup() {
+			return this.autoStartup;
+		}
+
+		public void setAutoStartup(boolean autoStartup) {
+			this.autoStartup = autoStartup;
+		}
+
+	}
+
+	public abstract static class AmqpContainer extends BaseContainer {
+
 		/**
 		 * Acknowledge mode of container.
 		 */
@@ -701,14 +732,6 @@ public abstract static class AmqpContainer {
 		 */
 		private final ListenerRetry retry = new ListenerRetry();
 
-		public boolean isAutoStartup() {
-			return this.autoStartup;
-		}
-
-		public void setAutoStartup(boolean autoStartup) {
-			this.autoStartup = autoStartup;
-		}
-
 		public AcknowledgeMode getAcknowledgeMode() {
 			return this.acknowledgeMode;
 		}
@@ -871,6 +894,24 @@ public void setMissingQueuesFatal(boolean missingQueuesFatal) {
 
 	}
 
+	public static class StreamContainer extends BaseContainer {
+
+		/**
+		 * When true, the container factory will create containers that support listeners
+		 * that consume native stream messages instead of spring-amqp {@code Message}s.
+		 */
+		boolean nativeListener;
+
+		public boolean isNativeListener() {
+			return this.nativeListener;
+		}
+
+		public void setNativeListener(boolean nativeListener) {
+			this.nativeListener = nativeListener;
+		}
+
+	}
+
 	public static class Template {
 
 		private final Retry retry = new Retry();
@@ -1128,4 +1169,62 @@ private boolean determineSslEnabled(boolean sslEnabled) {
 
 	}
 
+	public static final class Stream {
+
+		/**
+		 * Host of a RabbitMQ instance with the Stream Plugin Enabled
+		 */
+		private String host = "localhost";
+
+		/**
+		 * Stream port of a RabbitMQ instance with the Stream Plugin Enabled
+		 */
+		private int port = DEFAULT_STREAM_PORT;
+
+		/**
+		 * Login user to authenticate to the broker. If not set
+		 * {@code spring.rabbitmq.username} will be used.
+		 */
+		private String username;
+
+		/**
+		 * Login password to authenticate to the broker. If not set
+		 * {@code spring.rabbitmq.password} will be used.
+		 */
+		private String password;
+
+		public String getHost() {
+			return this.host;
+		}
+
+		public void setHost(String host) {
+			this.host = host;
+		}
+
+		public int getPort() {
+			return this.port;
+		}
+
+		public void setPort(int port) {
+			this.port = port;
+		}
+
+		public String getUsername() {
+			return this.username;
+		}
+
+		public void setUsername(String username) {
+			this.username = username;
+		}
+
+		public String getPassword() {
+			return this.password;
+		}
+
+		public void setPassword(String password) {
+			this.password = password;
+		}
+
+	}
+
 }

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitStreamConfiguration.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2021-2021 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.boot.autoconfigure.amqp;
+
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.EnvironmentBuilder;
+
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.config.ContainerCustomizer;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
+import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+
+/**
+ * Configuration for Spring RabbitMQ Stream Plugin support.
+ *
+ * @author Gary Russell
+ */
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnClass(StreamRabbitListenerContainerFactory.class)
+@ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "stream")
+class RabbitStreamConfiguration {
+
+	@Bean(name = "rabbitListenerContainerFactory")
+	@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
+	StreamRabbitListenerContainerFactory streamRabbitListenerContainerFactory(Environment rabbitStreamEnvironment,
+			RabbitProperties properties, ObjectProvider<ConsumerCustomizer> consumerCustomizer,
+			ObjectProvider<ContainerCustomizer<StreamListenerContainer>> containerCustomizer) {
+
+		StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(
+				rabbitStreamEnvironment);
+		factory.setNativeListener(properties.getListener().getStream().isNativeListener());
+		consumerCustomizer.ifUnique(factory::setConsumerCustomizer);
+		containerCustomizer.ifUnique(factory::setContainerCustomizer);
+		return factory;
+	}
+
+	@Bean(name = "rabbitStreamEnvironment")
+	@ConditionalOnMissingBean(name = "rabbitStreamEnvironment")
+	Environment rabbitStreamEnvironment(RabbitProperties properties) {
+		RabbitProperties.Stream stream = properties.getStream();
+		String username = stream.getUsername();
+		if (username == null) {
+			username = properties.getUsername();
+		}
+		String password = stream.getPassword();
+		if (password == null) {
+			password = properties.getPassword();
+		}
+		EnvironmentBuilder builder = Environment.builder().lazyInitialization(true).host(stream.getHost())
+				.port(stream.getPort());
+		if (username != null) {
+			builder.username(username);
+		}
+		if (password != null) {
+			builder.password(password);
+		}
+		return builder.build();
+	}
+
+}

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitStreamConfigurationTests.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2012-2021 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.boot.autoconfigure.amqp;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.config.ContainerCustomizer;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
+import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link RabbitStreamConfiguration}.
+ *
+ * @author Gary Russell
+ */
+class RabbitStreamConfigurationTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(RabbitAutoConfiguration.class));
+
+	@Test
+	void testContainerType() {
+		this.contextRunner.withUserConfiguration(TestConfiguration.class)
+				.withPropertyValues("spring.rabbitmq.listener.type:stream",
+						"spring.rabbitmq.listener.stream.native-listener:true")
+				.run((context) -> {
+					RabbitListenerEndpointRegistry registry = context.getBean(RabbitListenerEndpointRegistry.class);
+					assertThat(registry.getListenerContainer("test")).isInstanceOf(StreamListenerContainer.class);
+					assertThat(new DirectFieldAccessor(registry.getListenerContainer("test"))
+							.getPropertyValue("consumerCustomizer")).isNotNull();
+					assertThat(new DirectFieldAccessor(context.getBean(StreamRabbitListenerContainerFactory.class))
+							.getPropertyValue("nativeListener")).isEqualTo(Boolean.TRUE);
+					assertThat(context.getBean(TestConfiguration.class).containerCustomizerCalled).isTrue();
+				});
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableRabbit
+	static class TestConfiguration {
+
+		boolean containerCustomizerCalled;
+
+		@RabbitListener(id = "test", queues = "stream", autoStartup = "false")
+		void listen(String in) {
+		}
+
+		@Bean
+		ConsumerCustomizer consumerCustomizer() {
+			return (id, consumer) -> {
+			};
+		}
+
+		@Bean
+		ContainerCustomizer<StreamListenerContainer> containerCustomizer() {
+			return (container) -> this.containerCustomizerCalled = true;
+		}
+
+	}
+
+}

spring-boot-project/spring-boot-dependencies/build.gradle

@@ -1416,6 +1416,13 @@ bom {
 			]
 		}
 	}
+	library("Rabbit Stream Client", "0.3.0") {
+		group("com.rabbitmq") {
+			modules = [
+					"stream-client"
+			]
+		}
+	}
 	library("Reactive Streams", "1.0.3") {
 		group("org.reactivestreams") {
 			modules = [
@@ -1644,6 +1651,7 @@ bom {
 			modules = [
 				"spring-amqp",
 				"spring-rabbit",
+				"spring-rabbit-stream",
 				"spring-rabbit-junit",
 				"spring-rabbit-test"
 			]