Move logic of resolving PulsarBrokerUrl and HttpServiceUrl into Pulsa…

…rContainer
quarkusio · Nov 8, 2024 · 321efed · 321efed
1 parent 6f99bbc
commit 321efed
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 15 deletions.
diff --git a/...rc/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarContainer.java b/...rc/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarContainer.java
@@ -1,5 +1,7 @@
 package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
 
+import static io.quarkus.smallrye.reactivemessaging.pulsar.deployment.PulsarDevServicesProcessor.DEV_SERVICE_PULSAR;
+
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Collections;
@@ -11,6 +13,8 @@
 
 import com.github.dockerjava.api.command.InspectContainerResponse;
 
+import io.quarkus.devservices.common.ConfigureUtil;
+
 public class PulsarContainer extends GenericContainer<PulsarContainer> {
 
     public static final DockerImageName PULSAR_IMAGE = DockerImageName.parse("apachepulsar/pulsar:3.2.4");
@@ -20,6 +24,9 @@ public class PulsarContainer extends GenericContainer<PulsarContainer> {
     public static final int BROKER_PORT = 6650;
     public static final int BROKER_HTTP_PORT = 8080;
 
+    private boolean useSharedNetwork;
+    private String hostName;
+
     public PulsarContainer() {
         this(PULSAR_IMAGE);
     }
@@ -51,6 +58,13 @@ protected void containerIsStarting(InspectContainerResponse containerInfo, boole
                 STARTER_SCRIPT);
     }
 
+    public PulsarContainer withSharedNetwork() {
+        useSharedNetwork = true;
+        hostName = ConfigureUtil.configureSharedNetwork(this, DEV_SERVICE_PULSAR);
+
+        return self();
+    }
+
     public PulsarContainer withPort(final int fixedPort) {
         if (fixedPort <= 0) {
             throw new IllegalArgumentException("The fixed port must be greater than 0");
@@ -60,10 +74,24 @@ public PulsarContainer withPort(final int fixedPort) {
     }
 
     public String getPulsarBrokerUrl() {
-        return String.format("pulsar://%s:%s", this.getHost(), this.getMappedPort(BROKER_PORT));
+        if (useSharedNetwork) {
+            return getServiceUrl(this.hostName, PulsarContainer.BROKER_PORT);
+        }
+        return getServiceUrl(this.getHost(), this.getMappedPort(BROKER_PORT));
+    }
+
+    private String getServiceUrl(String host, int port) {
+        return String.format("pulsar://%s:%d", host, port);
     }
 
     public String getHttpServiceUrl() {
-        return String.format("http://%s:%s", this.getHost(), this.getMappedPort(BROKER_HTTP_PORT));
+        if (useSharedNetwork) {
+            return getHttpServiceUrl(this.hostName, PulsarContainer.BROKER_HTTP_PORT);
+        }
+        return getHttpServiceUrl(this.getHost(), this.getMappedPort(BROKER_HTTP_PORT));
+    }
+
+    private String getHttpServiceUrl(String host, int port) {
+        return String.format("http://%s:%d", host, port);
     }
 }
diff --git a/...a/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java b/...a/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java
@@ -29,7 +29,6 @@
 import io.quarkus.deployment.console.StartupLogCompressor;
 import io.quarkus.deployment.dev.devservices.GlobalDevServicesConfig;
 import io.quarkus.deployment.logging.LoggingSetupBuildItem;
-import io.quarkus.devservices.common.ConfigureUtil;
 import io.quarkus.devservices.common.ContainerLocator;
 import io.quarkus.runtime.LaunchMode;
 import io.quarkus.runtime.configuration.ConfigUtils;
@@ -51,9 +50,9 @@ public class PulsarDevServicesProcessor {
 
     private static final ContainerLocator pulsarContainerLocator = new ContainerLocator(DEV_SERVICE_LABEL,
             PulsarContainer.BROKER_PORT);
-    private static final String DEV_SERVICE_PULSAR = "pulsar";
     private static final String PULSAR_CLIENT_SERVICE_URL = "pulsar.client.serviceUrl";
     private static final String PULSAR_ADMIN_SERVICE_URL = "pulsar.admin.serviceUrl";
+    static final String DEV_SERVICE_PULSAR = "pulsar";
     static volatile RunningDevService devService;
     static volatile PulsarDevServiceCfg cfg;
     static volatile boolean first = true;
@@ -183,21 +182,13 @@ private RunningDevService startPulsarContainer(DockerStatusBuildItem dockerStatu
                 container.withPort(config.fixedExposedPort);
             }
             timeout.ifPresent(container::withStartupTimeout);
-            String hostName = null;
             if (useSharedNetwork) {
-                hostName = ConfigureUtil.configureSharedNetwork(container, DEV_SERVICE_PULSAR);
+                container.withSharedNetwork();
             }
             container.start();
 
-            var pulsarBrokerUrl = container.getPulsarBrokerUrl();
-            var httpHostServiceUrl = container.getHttpServiceUrl();
-            if (useSharedNetwork) {
-                pulsarBrokerUrl = getServiceUrl(hostName, PulsarContainer.BROKER_PORT);
-                httpHostServiceUrl = getHttpServiceUrl(hostName, PulsarContainer.BROKER_HTTP_PORT);
-            }
-
-            return getRunningService(container.getContainerId(), container::close, pulsarBrokerUrl,
-                    httpHostServiceUrl);
+            return getRunningService(container.getContainerId(), container::close, container.getPulsarBrokerUrl(),
+                    container.getHttpServiceUrl());
         };
 
         return pulsarContainerLocator.locateContainer(config.serviceName, config.shared, launchMode.getLaunchMode())