Graceful shutdown

pubsub-plus-connector/src/main/java/io/quarkiverse/solace/SolaceConnector.java

@@ -22,10 +22,8 @@
 
 import com.solace.messaging.MessagingService;
 
-import io.quarkiverse.solace.i18n.SolaceLogging;
 import io.quarkiverse.solace.incoming.SolaceIncomingChannel;
 import io.quarkiverse.solace.outgoing.SolaceOutgoingChannel;
-import io.quarkus.runtime.ShutdownEvent;
 import io.smallrye.reactive.messaging.annotations.ConnectorAttribute;
 import io.smallrye.reactive.messaging.connector.InboundConnector;
 import io.smallrye.reactive.messaging.connector.OutboundConnector;
@@ -50,8 +48,6 @@
 @ConnectorAttribute(name = "consumer.queue.replay.strategy", type = "string", direction = INCOMING, description = "The receiver replay strategy")
 @ConnectorAttribute(name = "consumer.queue.replay.timebased-start-time", type = "string", direction = INCOMING, description = "The receiver replay timebased start time")
 @ConnectorAttribute(name = "consumer.queue.replay.replication-group-message-id", type = "string", direction = INCOMING, description = "The receiver replay replication group message id")
-// TODO implement consumer concurrency
-//@ConnectorAttribute(name = "consumer.queue.concurrency", type = "int", direction = INCOMING, description = "The number of concurrent consumers", defaultValue = "1")
 @ConnectorAttribute(name = "consumer.queue.failure-strategy", type = "string", direction = INCOMING, description = "Specify the failure strategy to apply when a message consumed from Solace broker is nacked. Accepted values are `ignore` (default), `fail`, `discard`, `error_topic`.", defaultValue = "ignore")
 @ConnectorAttribute(name = "consumer.queue.error.topic", type = "string", direction = INCOMING, description = "The error topic where message should be published in case of error")
 @ConnectorAttribute(name = "consumer.queue.error.message.dmq-eligible", type = "boolean", direction = INCOMING, description = "Whether error message is eligible to move to dead message queue", defaultValue = "false")
@@ -81,18 +77,6 @@ public class SolaceConnector implements InboundConnector, OutboundConnector, Hea
     List<SolaceIncomingChannel> incomingChannels = new CopyOnWriteArrayList<>();
     List<SolaceOutgoingChannel> outgoingChannels = new CopyOnWriteArrayList<>();
 
-    public void onStop(@Observes ShutdownEvent shutdownEvent) {
-        if (solace.isConnected()) {
-            SolaceLogging.log.info("Waiting incoming channel messages to be acknowledged");
-            incomingChannels.forEach(SolaceIncomingChannel::waitForUnAcknowledgedMessages);
-            SolaceLogging.log.info("All incoming channel messages are acknowledged");
-
-            SolaceLogging.log.info("Waiting for outgoing messages to be published");
-            outgoingChannels.forEach(SolaceOutgoingChannel::waitForPublishedMessages);
-            SolaceLogging.log.info("All outgoing messages are published");
-        }
-    }
-
     public void terminate(
             @Observes(notifyObserver = Reception.IF_EXISTS) @Priority(50) @BeforeDestroyed(ApplicationScoped.class) Object event) {
         incomingChannels.forEach(SolaceIncomingChannel::close);

pubsub-plus-connector/src/main/java/io/quarkiverse/solace/incoming/SolaceIncomingChannel.java

@@ -173,6 +173,7 @@ public Flow.Publisher<? extends Message<?>> getStream() {
     public void waitForUnAcknowledgedMessages() {
         try {
             receiver.pause();
+            SolaceLogging.log.info("Waiting for incoming channel messages to be acknowledged");
             if (!unacknowledgedMessageTracker.awaitEmpty(this.waitTimeout, TimeUnit.MILLISECONDS)) {
                 SolaceLogging.log.info(String.format("Timed out while waiting for the" +
                         " remaining messages to be acknowledged."));
@@ -184,6 +185,7 @@ public void waitForUnAcknowledgedMessages() {
     }
 
     public void close() {
+        waitForUnAcknowledgedMessages();
         closed.compareAndSet(false, true);
         if (this.pollerThread != null) {
             this.pollerThread.shutdown();

pubsub-plus-connector/src/main/java/io/quarkiverse/solace/outgoing/SolaceOutgoingChannel.java

@@ -179,6 +179,7 @@ public Flow.Subscriber<? extends Message<?>> getSubscriber() {
 
     public void waitForPublishedMessages() {
         try {
+            SolaceLogging.log.info("Waiting for outgoing messages to be published");
             if (!publishedMessagesTracker.awaitEmpty(this.waitTimeout, TimeUnit.MILLISECONDS)) {
                 SolaceLogging.log.info(String.format("Timed out while waiting for the" +
                         " remaining messages to get publish acknowledgment."));
@@ -190,6 +191,7 @@ public void waitForPublishedMessages() {
     }
 
     public void close() {
+        waitForPublishedMessages();
         if (processor != null) {
             processor.cancel();
         }

pubsub-plus-connector/src/test/java/io/quarkiverse/solace/SolaceConsumerTest.java

@@ -4,14 +4,19 @@
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Flow;
+import java.util.concurrent.TimeUnit;
 
 import jakarta.enterprise.context.ApplicationScoped;
 
 import org.awaitility.Durations;
 import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Message;
 import org.junit.jupiter.api.MethodOrderer;
 import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
@@ -26,8 +31,12 @@
 
 import io.quarkiverse.solace.base.SolaceContainer;
 import io.quarkiverse.solace.base.WeldTestBase;
+import io.quarkiverse.solace.incoming.SolaceIncomingChannel;
 import io.quarkiverse.solace.logging.SolaceTestAppender;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.reactive.messaging.health.HealthReport;
 import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
+import io.vertx.mutiny.core.Vertx;
 
 @TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 public class SolaceConsumerTest extends WeldTestBase {
@@ -220,6 +229,49 @@ void consumerPublishToErrorTopicPermissionException() {
 
     @Test
     @Order(7)
+    void consumerGracefulCloseTest() {
+        MapBasedConfig config = new MapBasedConfig()
+                .with("channel-name", "in")
+                .with("consumer.queue.name", queue)
+                .with("consumer.queue.add-additional-subscriptions", true)
+                .with("consumer.queue.missing-resource-creation-strategy", "create-on-start")
+                .with("consumer.queue.subscriptions", topic);
+
+        // Initialize incoming channel to consumes messages
+        SolaceIncomingChannel solaceIncomingChannel = new SolaceIncomingChannel(Vertx.vertx(),
+                new SolaceConnectorIncomingConfiguration(config), messagingService);
+
+        List<Object> list = new ArrayList<>();
+
+        Flow.Publisher<? extends Message<?>> stream = solaceIncomingChannel.getStream();
+        Multi.createFrom().publisher(stream).subscribe().with(message -> {
+            list.add(message);
+            CompletableFuture.runAsync(message::ack);
+        });
+        await().until(() -> {
+            HealthReport.HealthReportBuilder builder = HealthReport.builder();
+            solaceIncomingChannel.isReady(builder);
+            return builder.build().isOk();
+        });
+
+        // Produce messages
+        PersistentMessagePublisher publisher = messagingService.createPersistentMessagePublisherBuilder()
+                .build()
+                .start();
+        Topic tp = Topic.of(topic);
+        publisher.publish("1", tp);
+        publisher.publish("2", tp);
+        publisher.publish("3", tp);
+        publisher.publish("4", tp);
+        publisher.publish("5", tp);
+
+        // Assert on consumed messages
+        await().atMost(2, TimeUnit.MINUTES).until(() -> list.size() == 5);
+        solaceIncomingChannel.close();
+    }
+
+    @Test
+    @Order(8)
     void consumerCreateMissingResourceAddSubscriptionPermissionException() {
         MapBasedConfig config = new MapBasedConfig()
                 .with("mp.messaging.incoming.in.connector", "quarkus-solace")

pubsub-plus-connector/src/test/java/io/quarkiverse/solace/SolacePublisherTest.java

@@ -6,6 +6,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Flow;
 
 import jakarta.enterprise.context.ApplicationScoped;
 
@@ -20,8 +21,11 @@
 
 import io.quarkiverse.solace.base.WeldTestBase;
 import io.quarkiverse.solace.outgoing.SolaceOutboundMetadata;
+import io.quarkiverse.solace.outgoing.SolaceOutgoingChannel;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.reactive.messaging.health.HealthReport;
 import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
+import io.vertx.mutiny.core.Vertx;
 
 public class SolacePublisherTest extends WeldTestBase {
 
@@ -99,6 +103,37 @@ void publisherWithBackPressureReject() {
         await().untilAsserted(() -> assertThat(app.getAcked().size()).isLessThan(5));
     }
 
+    @Test
+    void publisherGracefulCloseTest() {
+        MapBasedConfig config = new MapBasedConfig()
+                .with("channel-name", "out")
+                .with("producer.topic", topic);
+
+        List<String> expected = new CopyOnWriteArrayList<>();
+
+        // Start listening first
+        PersistentMessageReceiver receiver = messagingService.createPersistentMessageReceiverBuilder()
+                .withSubscriptions(TopicSubscription.of(topic))
+                .build(Queue.nonDurableExclusiveQueue());
+        receiver.receiveAsync(inboundMessage -> expected.add(inboundMessage.getPayloadAsString()));
+        receiver.start();
+
+        SolaceOutgoingChannel solaceOutgoingChannel = new SolaceOutgoingChannel(Vertx.vertx(),
+                new SolaceConnectorOutgoingConfiguration(config), messagingService);
+        // Publish messages
+        Multi.createFrom().range(0, 10)
+                .map(Message::of)
+                .subscribe((Flow.Subscriber<? super Message<Integer>>) solaceOutgoingChannel.getSubscriber());
+        await().until(() -> {
+            HealthReport.HealthReportBuilder builder = HealthReport.builder();
+            solaceOutgoingChannel.isReady(builder);
+            return builder.build().isOk();
+        });
+        // Assert on received messages
+        await().untilAsserted(() -> assertThat(expected.size()).isEqualTo(10));
+        solaceOutgoingChannel.close();
+    }
+
     //    @Test
     //    void publisherWithBackPressureRejectWaitForPublisherReadiness() {
     //        MapBasedConfig config = new MapBasedConfig()

runtime/src/main/java/io/quarkiverse/solace/runtime/SolaceRecorder.java

@@ -55,7 +55,7 @@ public MessagingService apply(SyntheticCreationalContext<MessagingService> conte
                 }
 
                 var tmp = service;
-                shutdown.addShutdownTask(() -> {
+                shutdown.addLastShutdownTask(() -> {
                     if (tmp.isConnected()) {
                         tmp.disconnect();
                     }