Add RabbitMQ Stream service connection from RabbitMQContainer

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

@@ -25,6 +25,7 @@
 import org.springframework.amqp.rabbit.config.ContainerCustomizer;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Stream;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties.StreamContainer;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -50,6 +51,12 @@
 @ConditionalOnClass(StreamRabbitListenerContainerFactory.class)
 class RabbitStreamConfiguration {
 
+	@Bean
+	@ConditionalOnMissingBean(RabbitStreamConnectionDetails.class)
+	RabbitStreamConnectionDetails rabbitStreamConnectionDetails(RabbitProperties rabbitProperties) {
+		return new PropertiesRabbitStreamConnectionDetails(rabbitProperties.getStream());
+	}
+
 	@Bean(name = "rabbitListenerContainerFactory")
 	@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
 	@ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "stream")
@@ -68,9 +75,9 @@ StreamRabbitListenerContainerFactory streamRabbitListenerContainerFactory(Enviro
 
 	@Bean(name = "rabbitStreamEnvironment")
 	@ConditionalOnMissingBean(name = "rabbitStreamEnvironment")
-	Environment rabbitStreamEnvironment(RabbitProperties properties,
+	Environment rabbitStreamEnvironment(RabbitProperties properties, RabbitStreamConnectionDetails connectionDetails,
 			ObjectProvider<EnvironmentBuilderCustomizer> customizers) {
-		EnvironmentBuilder builder = configure(Environment.builder(), properties);
+		EnvironmentBuilder builder = configure(Environment.builder(), properties, connectionDetails);
 		customizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
 		return builder.build();
 	}
@@ -99,23 +106,65 @@ RabbitStreamTemplate rabbitStreamTemplate(Environment rabbitStreamEnvironment, R
 		return template;
 	}
 
-	static EnvironmentBuilder configure(EnvironmentBuilder builder, RabbitProperties properties) {
+	static EnvironmentBuilder configure(EnvironmentBuilder builder, RabbitProperties properties,
+			RabbitStreamConnectionDetails connectionDetails) {
 		builder.lazyInitialization(true);
 		RabbitProperties.Stream stream = properties.getStream();
 		PropertyMapper map = PropertyMapper.get();
-		map.from(stream.getHost()).to(builder::host);
-		map.from(stream.getPort()).to(builder::port);
-		map.from(stream.getVirtualHost())
+		map.from(connectionDetails.getHost()).to(builder::host);
+		map.from(connectionDetails.getPort()).to(builder::port);
+		map.from(connectionDetails.getVirtualHost())
 			.as(withFallback(properties::getVirtualHost))
 			.whenNonNull()
 			.to(builder::virtualHost);
-		map.from(stream.getUsername()).as(withFallback(properties::getUsername)).whenNonNull().to(builder::username);
-		map.from(stream.getPassword()).as(withFallback(properties::getPassword)).whenNonNull().to(builder::password);
+		map.from(connectionDetails.getUsername())
+			.as(withFallback(properties::getUsername))
+			.whenNonNull()
+			.to(builder::username);
+		map.from(connectionDetails.getPassword())
+			.as(withFallback(properties::getPassword))
+			.whenNonNull()
+			.to(builder::password);
 		return builder;
 	}
 
 	private static Function<String, String> withFallback(Supplier<String> fallback) {
 		return (value) -> (value != null) ? value : fallback.get();
 	}
 
+	static class PropertiesRabbitStreamConnectionDetails implements RabbitStreamConnectionDetails {
+
+		private final Stream streamProperties;
+
+		PropertiesRabbitStreamConnectionDetails(Stream streamProperties) {
+			this.streamProperties = streamProperties;
+		}
+
+		@Override
+		public String getHost() {
+			return this.streamProperties.getHost();
+		}
+
+		@Override
+		public int getPort() {
+			return this.streamProperties.getPort();
+		}
+
+		@Override
+		public String getVirtualHost() {
+			return this.streamProperties.getVirtualHost();
+		}
+
+		@Override
+		public String getUsername() {
+			return this.streamProperties.getUsername();
+		}
+
+		@Override
+		public String getPassword() {
+			return this.streamProperties.getPassword();
+		}
+
+	}
+
 }

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

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2012-2024 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.springframework.boot.autoconfigure.service.connection.ConnectionDetails;
+
+/**
+ * Details required to establish a connection to a RabbitMQ Stream service.
+ *
+ * @author Eddú Meléndez
+ * @since 3.4.0
+ */
+public interface RabbitStreamConnectionDetails extends ConnectionDetails {
+
+	/**
+	 * Rabbit server host.
+	 * @return the rabbit server host
+	 */
+	String getHost();
+
+	/**
+	 * Rabbit Stream server port.
+	 * @return the rabbit stream server port
+	 */
+	int getPort();
+
+	/**
+	 * Login user to authenticate to the broker.
+	 * @return the login user to authenticate to the broker or {@code null}
+	 */
+	default String getUsername() {
+		return null;
+	}
+
+	/**
+	 * Login to authenticate against the broker.
+	 * @return the login to authenticate against the broker or {@code null}
+	 */
+	default String getPassword() {
+		return null;
+	}
+
+	/**
+	 * Virtual host to use when connecting to the broker.
+	 * @return the virtual host to use when connecting to the broker or {@code null}
+	 */
+	default String getVirtualHost() {
+		return null;
+	}
+
+}

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

@@ -17,7 +17,9 @@
 package org.springframework.boot.autoconfigure.amqp;
 
 import java.time.Duration;
+import java.util.List;
 
+import com.rabbitmq.stream.Address;
 import com.rabbitmq.stream.BackOffDelayPolicy;
 import com.rabbitmq.stream.Codec;
 import com.rabbitmq.stream.Environment;
@@ -32,6 +34,7 @@
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.amqp.RabbitStreamConfiguration.PropertiesRabbitStreamConnectionDetails;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -43,6 +46,7 @@
 import org.springframework.rabbit.stream.producer.ProducerCustomizer;
 import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
 import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.BDDMockito.then;
@@ -127,7 +131,7 @@ void whenCustomMessageListenerContainerFactoryIsDefinedThenAutoConfiguredContain
 	void environmentUsesPropertyDefaultsByDefault() {
 		EnvironmentBuilder builder = mock(EnvironmentBuilder.class);
 		RabbitProperties properties = new RabbitProperties();
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().port(5552);
 		then(builder).should().host("localhost");
 		then(builder).should().lazyInitialization(true);
@@ -141,7 +145,7 @@ void whenStreamPortIsSetThenEnvironmentUsesCustomPort() {
 		EnvironmentBuilder builder = mock(EnvironmentBuilder.class);
 		RabbitProperties properties = new RabbitProperties();
 		properties.getStream().setPort(5553);
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().port(5553);
 	}
 
@@ -150,7 +154,7 @@ void whenStreamHostIsSetThenEnvironmentUsesCustomHost() {
 		EnvironmentBuilder builder = mock(EnvironmentBuilder.class);
 		RabbitProperties properties = new RabbitProperties();
 		properties.getStream().setHost("stream.rabbit.example.com");
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().host("stream.rabbit.example.com");
 	}
 
@@ -159,7 +163,7 @@ void whenStreamVirtualHostIsSetThenEnvironmentUsesCustomVirtualHost() {
 		EnvironmentBuilder builder = mock(EnvironmentBuilder.class);
 		RabbitProperties properties = new RabbitProperties();
 		properties.getStream().setVirtualHost("stream-virtual-host");
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().virtualHost("stream-virtual-host");
 	}
 
@@ -168,7 +172,7 @@ void whenStreamVirtualHostIsNotSetButDefaultVirtualHostIsSetThenEnvironmentUsesD
 		EnvironmentBuilder builder = mock(EnvironmentBuilder.class);
 		RabbitProperties properties = new RabbitProperties();
 		properties.setVirtualHost("default-virtual-host");
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().virtualHost("default-virtual-host");
 	}
 
@@ -178,7 +182,7 @@ void whenStreamCredentialsAreNotSetThenEnvironmentUsesRabbitCredentials() {
 		RabbitProperties properties = new RabbitProperties();
 		properties.setUsername("alice");
 		properties.setPassword("secret");
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().username("alice");
 		then(builder).should().password("secret");
 	}
@@ -191,7 +195,7 @@ void whenStreamCredentialsAreSetThenEnvironmentUsesStreamCredentials() {
 		properties.setPassword("secret");
 		properties.getStream().setUsername("bob");
 		properties.getStream().setPassword("confidential");
-		RabbitStreamConfiguration.configure(builder, properties);
+		RabbitStreamConfiguration.configure(builder, properties, getRabbitConnectionDetails(properties));
 		then(builder).should().username("bob");
 		then(builder).should().password("confidential");
 	}
@@ -260,6 +264,22 @@ void environmentCreatedByBuilderCanBeCustomized() {
 		});
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	void connectionDetailsAreApplied() {
+		this.contextRunner.withPropertyValues("spring.rabbitmq.stream.name:stream-test")
+			.withUserConfiguration(CustomConnectionDetails.class)
+			.run((context) -> assertThat(context.getBean(Environment.class))
+				.extracting((environment) -> (List<Address>) ReflectionTestUtils.getField(environment, "addresses"))
+				.extracting((address) -> address.get(0))
+				.extracting("host", "port")
+				.containsExactly("rabbitmq", 5555));
+	}
+
+	private RabbitStreamConnectionDetails getRabbitConnectionDetails(RabbitProperties properties) {
+		return new PropertiesRabbitStreamConnectionDetails(properties.getStream());
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	static class TestConfiguration {
 
@@ -345,4 +365,24 @@ EnvironmentBuilderCustomizer customizerB() {
 
 	}
 
+	@Configuration(proxyBeanMethods = false)
+	static class CustomConnectionDetails {
+
+		@Bean
+		RabbitStreamConnectionDetails customRabbitMqStreamConnectionDetails() {
+			return new RabbitStreamConnectionDetails() {
+				@Override
+				public String getHost() {
+					return "rabbitmq";
+				}
+
+				@Override
+				public int getPort() {
+					return 5555;
+				}
+			};
+		}
+
+	}
+
 }

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

@@ -46,6 +46,7 @@ dependencies {
 	dockerTestImplementation("org.springframework:spring-jms")
 	dockerTestImplementation("org.springframework:spring-r2dbc")
 	dockerTestImplementation("org.springframework.amqp:spring-rabbit")
+	dockerTestImplementation("org.springframework.amqp:spring-rabbit-stream")
 	dockerTestImplementation("org.springframework.data:spring-data-redis")
 	dockerTestImplementation("org.springframework.kafka:spring-kafka")
 	dockerTestImplementation("org.springframework.ldap:spring-ldap-core")