From 321efede8ae69343774b7ce7e57eb1bf76332384 Mon Sep 17 00:00:00 2001 From: Jakub Gardo Date: Fri, 8 Nov 2024 23:32:44 +0100 Subject: [PATCH] Move logic of resolving PulsarBrokerUrl and HttpServiceUrl into PulsarContainer --- .../pulsar/deployment/PulsarContainer.java | 32 +++++++++++++++++-- .../PulsarDevServicesProcessor.java | 17 +++------- 2 files changed, 34 insertions(+), 15 deletions(-) diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarContainer.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarContainer.java index c21a5ac9f7263..fd3a4a94738e9 100644 --- a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarContainer.java +++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/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 { public static final DockerImageName PULSAR_IMAGE = DockerImageName.parse("apachepulsar/pulsar:3.2.4"); @@ -20,6 +24,9 @@ public class PulsarContainer extends GenericContainer { 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/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java index 507b0a3ea9e75..1c1184e534af0 100644 --- a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java +++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/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())