Rabbitmq addresses config for cluster-mode

smallrye-reactive-messaging-rabbitmq/src/main/java/io/smallrye/reactive/messaging/rabbitmq/RabbitMQClientHelper.java

@@ -8,11 +8,15 @@
 import static java.time.Duration.ofSeconds;
 
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Optional;
 
 import jakarta.enterprise.inject.Instance;
 import jakarta.enterprise.inject.literal.NamedLiteral;
 
+import com.rabbitmq.client.Address;
 import com.rabbitmq.client.impl.CredentialsProvider;
 import com.rabbitmq.client.impl.DefaultCredentialsRefreshService;
 
@@ -76,14 +80,14 @@ static RabbitMQOptions getClientOptions(Vertx vertx, RabbitMQConnectorCommonConf
         String connectionName = String.format("%s (%s)",
                 config.getChannel(),
                 config instanceof RabbitMQConnectorIncomingConfiguration ? "Incoming" : "Outgoing");
-        String host = config.getHost();
-        int port = config.getPort();
-        log.brokerConfigured(host, port, config.getChannel());
+        List<Address> addresses = config.getAddresses()
+                .map(s -> Arrays.asList(Address.parseAddresses(s)))
+                .orElseGet(() -> Collections.singletonList(new Address(config.getHost(), config.getPort())));
+        log.brokerConfigured(addresses.toString(), config.getChannel());
 
         RabbitMQOptions options = new RabbitMQOptions()
                 .setConnectionName(connectionName)
-                .setHost(host)
-                .setPort(port)
+                .setAddresses(addresses)
                 .setSsl(config.getSsl())
                 .setTrustAll(config.getTrustAll())
                 .setAutomaticRecoveryEnabled(config.getAutomaticRecoveryEnabled())

smallrye-reactive-messaging-rabbitmq/src/main/java/io/smallrye/reactive/messaging/rabbitmq/RabbitMQConnector.java

@@ -71,6 +71,7 @@
 @ConnectorAttribute(name = "password", direction = INCOMING_AND_OUTGOING, description = "The password used to authenticate to the broker", type = "string", alias = "rabbitmq-password")
 @ConnectorAttribute(name = "host", direction = INCOMING_AND_OUTGOING, description = "The broker hostname", type = "string", alias = "rabbitmq-host", defaultValue = "localhost")
 @ConnectorAttribute(name = "port", direction = INCOMING_AND_OUTGOING, description = "The broker port", type = "int", alias = "rabbitmq-port", defaultValue = "5672")
+@ConnectorAttribute(name = "addresses", direction = INCOMING_AND_OUTGOING, description = "The multiple addresses for cluster mode, when given overrides the host and port", type = "string", alias = "rabbitmq-addresses")
 @ConnectorAttribute(name = "ssl", direction = INCOMING_AND_OUTGOING, description = "Whether or not the connection should use SSL", type = "boolean", alias = "rabbitmq-ssl", defaultValue = "false")
 @ConnectorAttribute(name = "trust-all", direction = INCOMING_AND_OUTGOING, description = "Whether to skip trust certificate verification", type = "boolean", alias = "rabbitmq-trust-all", defaultValue = "false")
 @ConnectorAttribute(name = "trust-store-path", direction = INCOMING_AND_OUTGOING, description = "The path to a JKS trust store", type = "string", alias = "rabbitmq-trust-store-path")

smallrye-reactive-messaging-rabbitmq/src/main/java/io/smallrye/reactive/messaging/rabbitmq/i18n/RabbitMQLogging.java

@@ -115,8 +115,8 @@ public interface RabbitMQLogging extends BasicLogger {
     void createClientFromBean(String optionsBeanName);
 
     @LogMessage(level = Logger.Level.INFO)
-    @Message(id = 17036, value = "RabbitMQ broker configured to %s:%d for channel %s")
-    void brokerConfigured(String host, int port, String channel);
+    @Message(id = 17036, value = "RabbitMQ broker configured to %s for channel %s")
+    void brokerConfigured(String address, String channel);
 
     @LogMessage(level = Logger.Level.ERROR)
     @Message(id = 17037, value = "Unable to create client")