Some tests and cleanup

Author: ozangunalp

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

@@ -24,18 +24,40 @@ public class ClientHolder {
     private final AtomicReference<CurrentConnection> connectionHolder = new AtomicReference<>();
     private final AtomicReference<Context> rootContext;
     private final AtomicReference<CompletionStage<RabbitMQClient>> connectionStage = new AtomicReference<>();
+    private final Uni<RabbitMQClient> connection;
 
     private final Vertx vertx;
-    private final RabbitMQConnectorCommonConfiguration configuration;
 
     public ClientHolder(RabbitMQClient client,
-            RabbitMQConnectorCommonConfiguration configuration,
+            String channel,
             Vertx vertx,
             Context root) {
         this.client = client;
-        this.configuration = configuration;
         this.vertx = vertx;
         this.rootContext = new AtomicReference<>(root);
+        this.connection = Uni.createFrom().deferred(() -> client.start()
+                .onSubscription().invoke(() -> {
+                    connected.set(true);
+                    log.connectionEstablished(channel);
+                })
+                .onItem().transform(ignored -> {
+                    Context ctx = rootContext.get();
+                    connectionHolder
+                            .set(new CurrentConnection(client, ctx == null ? Vertx.currentContext() : ctx));
+
+                    // handle the case we are already disconnected.
+                    if (!client.isConnected() || connectionHolder.get() == null) {
+                        // Throwing the exception would trigger a retry.
+                        connectionHolder.set(null);
+                        throw ex.illegalStateConnectionDisconnected();
+                    }
+                    return client;
+                })
+                .onFailure().invoke(log::unableToConnectToBroker)
+                .onFailure().invoke(t -> {
+                    connectionHolder.set(null);
+                    log.unableToRecoverFromConnectionDisruption(t);
+                }));
     }
 
     public static CompletionStage<Void> runOnContext(Context context, IncomingRabbitMQMessage<?> msg,
@@ -110,7 +132,7 @@ public Uni<RabbitMQClient> getOrEstablishConnection() {
             if (current != null) {
                 return Uni.createFrom().completionStage(current);
             }
-            CompletionStage<RabbitMQClient> created = createConnectionUni().subscribeAsCompletionStage();
+            CompletionStage<RabbitMQClient> created = connection.subscribeAsCompletionStage();
             if (connectionStage.compareAndSet(null, created)) {
                 created.whenComplete((result, error) -> {
                     if (error != null) {
@@ -133,32 +155,4 @@ private CurrentConnection(RabbitMQClient client, Context context) {
         }
     }
 
-    private Uni<RabbitMQClient> createConnectionUni() {
-        return Uni.createFrom().deferred(() -> client.start()
-                .onSubscription().invoke(() -> {
-                    connected.set(true);
-                    log.connectionEstablished(configuration.getChannel());
-                })
-                .onItem().transform(ignored -> {
-                    Context context = rootContext.get();
-                    if (context == null) {
-                        context = Vertx.currentContext();
-                    }
-                    connectionHolder.set(new CurrentConnection(client, context));
-
-                    // handle the case we are already disconnected.
-                    if (!client.isConnected() || connectionHolder.get() == null) {
-                        // Throwing the exception would trigger a retry.
-                        connectionHolder.set(null);
-                        throw ex.illegalStateConnectionDisconnected();
-                    }
-                    return client;
-                })
-                .onFailure().invoke(log::unableToConnectToBroker)
-                .onFailure().invoke(t -> {
-                    connectionHolder.set(null);
-                    log.unableToRecoverFromConnectionDisruption(t);
-                }));
-    }
-
 }

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

@@ -72,11 +72,6 @@ public IncomingRabbitMQMessage(RabbitMQMessage delegate, ClientHolder holder, Co
         this(delegate.getDelegate(), holder, context, onNack, onAck, contentTypeOverride);
     }
 
-    IncomingRabbitMQMessage(io.vertx.rabbitmq.RabbitMQMessage msg, ClientHolder holder,
-            RabbitMQFailureHandler onNack, RabbitMQAckHandler onAck, String contentTypeOverride) {
-        this(msg, holder, holder.getContext(), onNack, onAck, contentTypeOverride);
-    }
-
     IncomingRabbitMQMessage(io.vertx.rabbitmq.RabbitMQMessage msg, ClientHolder holder, Context context,
             RabbitMQFailureHandler onNack, RabbitMQAckHandler onAck, String contentTypeOverride) {
         this.message = msg;

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

@@ -329,7 +329,8 @@ public ClientHolder getClientHolder(RabbitMQConnectorCommonConfiguration config,
 
     private ClientHolder createAndRegisterHolder(RabbitMQConnectorCommonConfiguration config,
             io.vertx.mutiny.core.Context context, String key, boolean shared) {
-        ClientHolder holder = new ClientHolder(RabbitMQClientHelper.createClient(this, config), config, vertx(), context);
+        ClientHolder holder = new ClientHolder(RabbitMQClientHelper.createClient(this, config), config.getChannel(), vertx(),
+                context);
         clientRegistrations.put(config.getChannel(), new ClientRegistration(holder, shared, key));
         return holder;
     }
@@ -351,7 +352,7 @@ private ClientHolder getOrCreateSharedHolder(RabbitMQConnectorCommonConfiguratio
             }
             return new SharedClient(name, new ClientHolder(
                     RabbitMQClient.create(vertx(), options),
-                    config,
+                    config.getChannel(),
                     vertx(),
                     context), fingerprint);
         });

smallrye-reactive-messaging-rabbitmq/src/main/java/io/smallrye/reactive/messaging/rabbitmq/internals/OutgoingRabbitMQChannel.java

@@ -26,6 +26,7 @@ public class OutgoingRabbitMQChannel {
     private final Flow.Subscriber<Message<?>> subscriber;
     private final RabbitMQConnectorOutgoingConfiguration config;
     private final ClientHolder holder;
+    private final RabbitMQMessageSender processor;
     private volatile RabbitMQPublisher publisher;
 
     public OutgoingRabbitMQChannel(RabbitMQConnector connector, RabbitMQConnectorOutgoingConfiguration oc,
@@ -54,10 +55,19 @@ public OutgoingRabbitMQChannel(RabbitMQConnector connector, RabbitMQConnectorOut
                 .onFailure().recoverWithNull().memoize().indefinitely();
 
         // Set up a sender based on the publisher we established above
-        final RabbitMQMessageSender processor = new RabbitMQMessageSender(oc, getSender, openTelemetryInstance);
+        processor = new RabbitMQMessageSender(oc, getSender, openTelemetryInstance);
 
         // Return a SubscriberBuilder
-        subscriber = MultiUtils.via(processor, m -> m.onFailure().invoke(t -> log.error(oc.getChannel(), t)));
+        subscriber = MultiUtils.via(processor, m -> m.onFailure().invoke(t -> log.error(oc.getChannel(), t))
+                .onTermination().call(() -> {
+                    if (publisher != null) {
+                        return publisher.stop()
+                                .ifNoItem().after(Duration.ofSeconds(oc.getReconnectInterval())).fail()
+                                .onFailure()
+                                .invoke(e -> log.infof(e, "Error terminating outgoing channel %s", config.getChannel()));
+                    }
+                    return Uni.createFrom().voidItem();
+                }));
     }
 
     public Flow.Subscriber<Message<?>> getSubscriber() {
@@ -95,12 +105,6 @@ public HealthReport.HealthReportBuilder isReady(HealthReport.HealthReportBuilder
     }
 
     public void terminate() {
-        if (publisher != null) {
-            try {
-                publisher.stop().await().atMost(Duration.ofMillis(config.getConnectionTimeout()));
-            } catch (Exception e) {
-                log.infof(e, "Error terminating outgoing channel %s", config.getChannel());
-            }
-        }
+        processor.cancel();
     }
 }

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

@@ -32,11 +32,6 @@
 import io.smallrye.reactive.messaging.rabbitmq.RabbitMQConnectorIncomingConfiguration;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.JksOptions;
-import io.vertx.core.net.KeyCertOptions;
-import io.vertx.core.net.PemKeyCertOptions;
-import io.vertx.core.net.PemTrustOptions;
-import io.vertx.core.net.PfxOptions;
-import io.vertx.core.net.TrustOptions;
 import io.vertx.mutiny.core.Vertx;
 import io.vertx.mutiny.rabbitmq.RabbitMQClient;
 import io.vertx.rabbitmq.RabbitMQOptions;
@@ -80,47 +75,12 @@ public static RabbitMQOptions buildClientOptions(RabbitMQConnector connector, Ra
     }
 
     public static String computeConnectionFingerprint(RabbitMQOptions options) {
-        StringBuilder raw = new StringBuilder();
-        append(raw, "uri", options.getUri());
-
+        JsonObject json = options.toJson();
         List<Address> addresses = options.getAddresses();
-        if (addresses != null && !addresses.isEmpty()) {
-            List<String> normalized = addresses.stream()
-                    .map(address -> address.getHost() + ":" + address.getPort())
-                    .sorted()
-                    .collect(Collectors.toList());
-            append(raw, "addresses", String.join(",", normalized));
-        } else {
-            append(raw, "host", options.getHost());
-            append(raw, "port", Integer.toString(options.getPort()));
+        if (addresses != null) {
+            json.put("addresses", addresses.stream().map(Address::toString).collect(Collectors.toList()));
         }
-
-        append(raw, "virtualHost", options.getVirtualHost());
-        append(raw, "user", options.getUser());
-        append(raw, "passwordHash", hashValue(options.getPassword()));
-
-        append(raw, "ssl", Boolean.toString(options.isSsl()));
-        append(raw, "trustAll", Boolean.toString(options.isTrustAll()));
-        append(raw, "hostnameVerificationAlgorithm", options.getHostnameVerificationAlgorithm());
-        append(raw, "keyCertOptions", keyCertFingerprint(options.getKeyCertOptions()));
-        append(raw, "trustOptions", trustFingerprint(options.getTrustOptions()));
-
-        append(raw, "connectionTimeout", Integer.toString(options.getConnectionTimeout()));
-        append(raw, "handshakeTimeout", Integer.toString(options.getHandshakeTimeout()));
-        append(raw, "requestedHeartbeat", Integer.toString(options.getRequestedHeartbeat()));
-        append(raw, "requestedChannelMax", Integer.toString(options.getRequestedChannelMax()));
-        append(raw, "networkRecoveryInterval", Long.toString(options.getNetworkRecoveryInterval()));
-        append(raw, "automaticRecoveryEnabled", Boolean.toString(options.isAutomaticRecoveryEnabled()));
-        append(raw, "automaticRecoveryOnInitialConnection", Boolean.toString(options.isAutomaticRecoveryOnInitialConnection()));
-        append(raw, "useNio", Boolean.toString(options.isNioEnabled()));
-        append(raw, "reconnectAttempts", Integer.toString(options.getReconnectAttempts()));
-        append(raw, "reconnectInterval", Long.toString(options.getReconnectInterval()));
-
-        append(raw, "credentialsProvider", className(options.getCredentialsProvider()));
-        append(raw, "credentialsRefreshService", className(options.getCredentialsRefreshService()));
-        append(raw, "saslConfig", className(options.getSaslConfig()));
-
-        return sha256(raw.toString());
+        return sha256(json.encode());
     }
 
     static RabbitMQOptions getClientOptionsFromBean(Instance<RabbitMQOptions> options, String optionsBeanName) {
@@ -223,21 +183,6 @@ private static String resolveConnectionName(RabbitMQConnectorCommonConfiguration
                         config instanceof RabbitMQConnectorIncomingConfiguration ? "Incoming" : "Outgoing"));
     }
 
-    private static void append(StringBuilder target, String key, String value) {
-        target.append(key).append('=').append(value == null ? "" : value).append(';');
-    }
-
-    private static String className(Object value) {
-        return value == null ? "" : value.getClass().getName();
-    }
-
-    private static String hashValue(String value) {
-        if (value == null) {
-            return "";
-        }
-        return sha256(value);
-    }
-
     private static String sha256(String value) {
         try {
             MessageDigest digest = MessageDigest.getInstance("SHA-256");
@@ -252,48 +197,6 @@ private static String sha256(String value) {
         }
     }
 
-    private static String keyCertFingerprint(KeyCertOptions options) {
-        if (options == null) {
-            return "";
-        }
-        if (options instanceof JksOptions) {
-            JksOptions jks = (JksOptions) options;
-            return String.join(":", "JKS", nullToEmpty(jks.getPath()), nullToEmpty(jks.getAlias()));
-        }
-        if (options instanceof PfxOptions) {
-            PfxOptions pfx = (PfxOptions) options;
-            return String.join(":", "PFX", nullToEmpty(pfx.getPath()), nullToEmpty(pfx.getAlias()));
-        }
-        if (options instanceof PemKeyCertOptions) {
-            PemKeyCertOptions pem = (PemKeyCertOptions) options;
-            return String.join(":", "PEM", String.join(",", pem.getKeyPaths()), String.join(",", pem.getCertPaths()));
-        }
-        return options.getClass().getName();
-    }
-
-    private static String trustFingerprint(TrustOptions options) {
-        if (options == null) {
-            return "";
-        }
-        if (options instanceof JksOptions) {
-            JksOptions jks = (JksOptions) options;
-            return String.join(":", "JKS", nullToEmpty(jks.getPath()), nullToEmpty(jks.getAlias()));
-        }
-        if (options instanceof PfxOptions) {
-            PfxOptions pfx = (PfxOptions) options;
-            return String.join(":", "PFX", nullToEmpty(pfx.getPath()), nullToEmpty(pfx.getAlias()));
-        }
-        if (options instanceof PemTrustOptions) {
-            PemTrustOptions pem = (PemTrustOptions) options;
-            return String.join(":", "PEM", String.join(",", pem.getCertPaths()));
-        }
-        return options.getClass().getName();
-    }
-
-    private static String nullToEmpty(String value) {
-        return value == null ? "" : value;
-    }
-
     public static String serverQueueName(String name) {
         if (name.equals("(server.auto)")) {
             return "";

smallrye-reactive-messaging-rabbitmq/src/test/java/io/smallrye/reactive/messaging/rabbitmq/DualIncomingContextBean.java

@@ -0,0 +1,69 @@
+package io.smallrye.reactive.messaging.rabbitmq;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+import io.smallrye.mutiny.Uni;
+import io.smallrye.reactive.messaging.providers.locals.LocalContextMetadata;
+import io.vertx.core.Context;
+
+@ApplicationScoped
+public class DualIncomingContextBean {
+
+    private final CountDownLatch latch1 = new CountDownLatch(1);
+    private final CountDownLatch latch2 = new CountDownLatch(1);
+    private final AtomicReference<Context> context1 = new AtomicReference<>();
+    private final AtomicReference<Context> context2 = new AtomicReference<>();
+    private final AtomicReference<Boolean> eventLoop1 = new AtomicReference<>();
+    private final AtomicReference<Boolean> eventLoop2 = new AtomicReference<>();
+
+    @Incoming("data1")
+    public Uni<Void> consume1(Message<Integer> message) {
+        message.getMetadata(LocalContextMetadata.class).ifPresent(metadata -> {
+            Context context = metadata.context();
+            context1.set(context);
+            eventLoop1.set(context.isEventLoopContext());
+        });
+        latch1.countDown();
+        return Uni.createFrom().voidItem();
+    }
+
+    @Incoming("data2")
+    public Uni<Void> consume2(Message<Integer> message) {
+        message.getMetadata(LocalContextMetadata.class).ifPresent(metadata -> {
+            Context context = metadata.context();
+            context2.set(context);
+            eventLoop2.set(context.isEventLoopContext());
+        });
+        latch2.countDown();
+        return Uni.createFrom().voidItem();
+    }
+
+    public boolean awaitMessages(long timeout, TimeUnit unit) throws InterruptedException {
+        return latch1.await(timeout, unit) && latch2.await(timeout, unit);
+    }
+
+    public Context getContext1() {
+        return context1.get();
+    }
+
+    public Context getContext2() {
+        return context2.get();
+    }
+
+    public boolean isEventLoop1() {
+        Boolean value = eventLoop1.get();
+        return value != null && value;
+    }
+
+    public boolean isEventLoop2() {
+        Boolean value = eventLoop2.get();
+        return value != null && value;
+    }
+}

smallrye-reactive-messaging-rabbitmq/src/test/java/io/smallrye/reactive/messaging/rabbitmq/OutgoingBean.java

@@ -2,9 +2,10 @@
 
 import jakarta.enterprise.context.ApplicationScoped;
 
-import org.eclipse.microprofile.reactive.messaging.Message;
 import org.eclipse.microprofile.reactive.messaging.Outgoing;
 
+import io.smallrye.mutiny.Multi;
+
 /**
  * A bean that can be registered to do just enough to support the
  * declaration of an exchange backing an outgoing rabbitmq channel.
@@ -13,8 +14,8 @@
 public class OutgoingBean {
 
     @Outgoing("sink")
-    public Message<String> process() {
-        return Message.of("test");
+    public Multi<String> process() {
+        return Multi.createFrom().items("test", "test2", "test3");
     }
 
 }