Enabled Rabbit Streams by adding ability to expose additional ports

embedded-rabbitmq/README.adoc

@@ -21,8 +21,9 @@
 * `embedded.rabbitmq.dockerImage` `(default is 'rabbitmq:4.0-alpine')`
 ** Image versions on https://hub.docker.com/_/rabbitmq?tab=tags[dockerhub]
 * `embedded.rabbitmq.waitTimeoutInSeconds` `(default is 60 seconds)`
-* `embedded.rabbitmq.enabledPlugins` `(list of strings, e.g. `rabbitmq_consistent_hash_exchange`, defaults to an empty list)`
+* `embedded.rabbitmq.enabledPlugins` (list of strings, e.g. `rabbitmq_consistent_hash_exchange`; defaults to an empty list)`
 * `embedded.toxiproxy.proxies.rabbitmq.enabled` Enables both creation of the container with ToxiProxy TCP proxy and a proxy to the `embedded-rabbitmq` container.
+* `embedded.rabbitmq.additionalPorts` (list of integers to expose from the container, e.g. `5552`; defaults to an empty list)`
 
 
 ==== Produces
@@ -39,5 +40,24 @@
 * `embedded.rabbitmq.internalPort`
 * `embedded.rabbitmq.internalHttpPort`
 * Bean `ToxiproxyContainer.ContainerProxy rabbitmqContainerProxy`
+* `embedded.rabbitmq.additionalPorts.x` (where `x` is the port from the container, e.g. `5552`; the value is the mapped port within the Docker container)
 
-//TODO: example missing
+==== Example
+
+===== RabbitMQ Streams
+To enable the RabbitMQ streams plugin and expose port 5552 (the Streams default protocol port) set:
+
+./src/test/resources/bootstrap.properties
+----
+embedded.rabbitmq.enabled-plugins[0]=rabbitmq_stream
+embedded.rabbitmq.additionalPorts[0]=5552
+----
+
+./src/test/resources/application.properties
+----
+spring.rabbitmq.stream.host=${embedded.rabbitmq.host}
+spring.rabbitmq.stream.port=${embedded.rabbitmq.additionalPorts.5552}
+spring.rabbitmq.stream.username=${embedded.rabbitmq.user}
+spring.rabbitmq.stream.password=${embedded.rabbitmq.password}
+spring.rabbitmq.stream.virtual-host=${embedded.rabbitmq.vhost}
+----

embedded-rabbitmq/pom.xml

@@ -31,5 +31,10 @@
             <artifactId>spring-rabbit</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.amqp</groupId>
+            <artifactId>spring-rabbit-stream</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

embedded-rabbitmq/src/main/java/com/playtika/testcontainer/rabbitmq/EmbeddedRabbitMQBootstrapConfiguration.java

@@ -20,6 +20,7 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import static com.playtika.testcontainer.common.utils.ContainerUtils.configureCommonsAndStart;
 import static com.playtika.testcontainer.rabbitmq.RabbitMQProperties.BEAN_NAME_EMBEDDED_RABBITMQ;
@@ -58,11 +59,16 @@ ToxiproxyContainer.ContainerProxy rabbitmqContainerProxy(ToxiproxyContainer toxi
     public RabbitMQContainer rabbitmq(ConfigurableEnvironment environment,
                                       RabbitMQProperties properties,
                                       Optional<Network> network) {
+
+        Integer[] exposedPorts = Stream.concat(properties.getAdditionalPorts().stream(), Stream.of(properties.getPort(), properties.getHttpPort()))
+                .distinct()
+                .toArray(Integer[]::new);
+
         RabbitMQContainer rabbitMQ =
                 new RabbitMQContainer(ContainerUtils.getDockerImageName(properties))
                         .withAdminPassword(properties.getPassword())
                         .withEnv("RABBITMQ_DEFAULT_VHOST", properties.getVhost())
-                        .withExposedPorts(properties.getPort(), properties.getHttpPort())
+                        .withExposedPorts(exposedPorts)
                         .withNetworkAliases(RABBITMQ_NETWORK_ALIAS);
 
         if (properties.getEnabledPlugins() != null && properties.getEnabledPlugins().size() != 0) {
@@ -80,6 +86,12 @@ private void registerRabbitMQEnvironment(RabbitMQContainer rabbitMQ,
                                              RabbitMQProperties properties) {
         Integer mappedPort = rabbitMQ.getMappedPort(properties.getPort());
         Integer mappedHttpPort = rabbitMQ.getMappedPort(properties.getHttpPort());
+
+        Map<Integer, Integer> additionalPorts = new LinkedHashMap<>();
+        for (Integer port : properties.getAdditionalPorts()) {
+            additionalPorts.put(port, rabbitMQ.getMappedPort(port));
+        }
+
         String host = rabbitMQ.getHost();
 
         LinkedHashMap<String, Object> map = new LinkedHashMap<>();
@@ -92,6 +104,9 @@ private void registerRabbitMQEnvironment(RabbitMQContainer rabbitMQ,
         map.put("embedded.rabbitmq.networkAlias", RABBITMQ_NETWORK_ALIAS);
         map.put("embedded.rabbitmq.internalPort", properties.getPort());
         map.put("embedded.rabbitmq.internalHttpPort", properties.getHttpPort());
+        for (Integer port : additionalPorts.keySet()) {
+            map.put("embedded.rabbitmq.additionalPorts." + port, additionalPorts.get(port));
+        }
 
         log.info("Started RabbitMQ server. Connection details: {}", map);
 

embedded-rabbitmq/src/test/java/com/playtika/testcontainer/rabbitmq/EmbeddedRabbitMQAdditionalPortsTest.java

@@ -0,0 +1,32 @@
+package com.playtika.testcontainer.rabbitmq;
+
+import com.rabbitmq.stream.Environment;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ActiveProfiles;
+
+@Slf4j
+@SpringBootTest(
+        classes = EmbeddedRabbitMQAdditionalPortsTest.TestConfiguration.class
+)
+@ActiveProfiles({"enabled", "stream"})
+public class EmbeddedRabbitMQAdditionalPortsTest {
+
+    @EnableAutoConfiguration
+    @Configuration
+    static class TestConfiguration {
+    }
+
+    @Autowired
+    Environment environment;
+
+    @Test
+    void streamPortExposed() {
+        environment.streamCreator().name("stream").create();
+        environment.streamExists("stream");
+    }
+}

embedded-rabbitmq/src/test/resources/application-enabled.yml

@@ -0,0 +1,7 @@
+spring:
+  rabbitmq:
+    host: ${embedded.rabbitmq.host}
+    port: ${embedded.rabbitmq.port}
+    username: ${embedded.rabbitmq.user}
+    password: ${embedded.rabbitmq.password}
+    virtual-host: ${embedded.rabbitmq.vhost}

embedded-rabbitmq/src/test/resources/application-stream.yml

@@ -0,0 +1,8 @@
+spring:
+  rabbitmq:
+    stream:
+      host: ${embedded.rabbitmq.host}
+      port: ${embedded.rabbitmq.additionalPorts.5552}
+      username: ${embedded.rabbitmq.user}
+      password: ${embedded.rabbitmq.password}
+      virtual-host: ${embedded.rabbitmq.vhost}

embedded-rabbitmq/src/test/resources/bootstrap-plugins.yml

@@ -0,0 +1,4 @@
+embedded:
+  rabbitmq:
+    enabled-plugins:
+      - rabbitmq_consistent_hash_exchange

embedded-rabbitmq/src/test/resources/bootstrap-stream.yml

@@ -0,0 +1,6 @@
+embedded:
+  rabbitmq:
+    enabled-plugins:
+      - rabbitmq_stream
+    additionalPorts:
+      - 5552

testcontainers-common/src/main/java/com/playtika/testcontainer/common/properties/CommonContainerProperties.java

@@ -77,6 +77,8 @@ public abstract class CommonContainerProperties {
     @Valid
     private List<MountVolume> mountVolumes = new ArrayList<>();
 
+    private List<Integer> additionalPorts = new ArrayList<>();
+
     /**
      * The Linux capabilities that should be enabled.
      */