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,12 +106,13 @@ 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(connectionDetails.getHost()).to(builder::host);
+		map.from(connectionDetails.getPort()).to(builder::port);
 		map.from(stream.getVirtualHost())
 			.as(withFallback(properties::getVirtualHost))
 			.whenNonNull()
@@ -118,4 +126,24 @@ 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();
+		}
+
+	}
+
 }

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

@@ -0,0 +1,41 @@
+/*
+ * 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 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 server port.
+	 * @return the rabbit server port
+	 */
+	int getPort();
+
+}

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")

spring-boot-project/spring-boot-testcontainers/src/dockerTest/java/org/springframework/boot/testcontainers/service/connection/amqp/RabbitStreamContainerConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,129 @@
+/*
+ * 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.testcontainers.service.connection.amqp;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.rabbitmq.stream.Address;
+import com.rabbitmq.stream.Environment;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.RabbitMQContainer;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.autoconfigure.amqp.EnvironmentBuilderCustomizer;
+import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
+import org.springframework.boot.autoconfigure.amqp.RabbitStreamConnectionDetails;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.testsupport.container.TestImage;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
+import org.springframework.rabbit.stream.support.StreamAdmin;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link RabbitStreamContainerConnectionDetailsFactory}.
+ *
+ * @author Eddú Meléndez
+ */
+@SpringJUnitConfig
+@TestPropertySource(
+		properties = { "spring.rabbitmq.stream.name=stream.queue1", "spring.rabbitmq.listener.type=stream" })
+@Testcontainers(disabledWithoutDocker = true)
+class RabbitStreamContainerConnectionDetailsFactoryIntegrationTests {
+
+	private static final int RABBITMQ_STREAMS_PORT = 5552;
+
+	@Container
+	@ServiceConnection
+	static final RabbitMQContainer rabbit = getRabbitMqStreamContainer();
+
+	private static RabbitMQContainer getRabbitMqStreamContainer() {
+		RabbitMQContainer container = TestImage.container(RabbitMQContainer.class);
+		container.addExposedPorts(RABBITMQ_STREAMS_PORT);
+		var enabledPlugins = "[rabbitmq_stream,rabbitmq_prometheus].";
+		container.withCopyToContainer(Transferable.of(enabledPlugins), "/etc/rabbitmq/enabled_plugins");
+		return container;
+	}
+
+	@Autowired(required = false)
+	private RabbitStreamConnectionDetails connectionDetails;
+
+	@Autowired
+	private RabbitStreamTemplate rabbitStreamTemplate;
+
+	@Autowired
+	private TestListener listener;
+
+	@Test
+	void connectionCanBeMadeToRabbitContainer() {
+		assertThat(this.connectionDetails).isNotNull();
+		this.rabbitStreamTemplate.convertAndSend("message");
+		Awaitility.waitAtMost(Duration.ofMinutes(4))
+			.untilAsserted(() -> assertThat(this.listener.messages).containsExactly("message"));
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ImportAutoConfiguration(RabbitAutoConfiguration.class)
+	static class TestConfiguration {
+
+		@Bean
+		StreamAdmin streamAdmin(Environment env) {
+			return new StreamAdmin(env, sc -> {
+				sc.stream("stream.queue1").create();
+			});
+		}
+
+		@Bean
+		EnvironmentBuilderCustomizer environmentBuilderCustomizer() {
+			return env -> {
+				Address entrypoint = new Address(rabbit.getHost(), rabbit.getMappedPort(RABBITMQ_STREAMS_PORT));
+				env.addressResolver(address -> entrypoint);
+			};
+		}
+
+		@Bean
+		TestListener testListener() {
+			return new TestListener();
+		}
+
+	}
+
+	static class TestListener {
+
+		private final List<String> messages = new ArrayList<>();
+
+		@RabbitListener(queues = "stream.queue1")
+		void processMessage(String message) {
+			this.messages.add(message);
+		}
+
+	}
+
+}