Fixed not reporting health down when not able to subscripe to NATS

.github/project.yml

@@ -1,4 +1,4 @@
 release:
-  current-version: "3.15.5"
+  current-version: "3.15.6"
   next-version: "3.16.0-SNAPSHOT"
 

deployment/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/deployment/ReactiveMesssagingNatsJetstreamProcessor.java

@@ -11,7 +11,7 @@
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultMessageMapper;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultPayloadMapper;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.StreamStateMapperImpl;
-import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrumenter;
+import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrument;
 import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
 import io.quarkus.deployment.annotations.BuildProducer;
 import io.quarkus.deployment.annotations.BuildStep;
@@ -51,7 +51,7 @@ void initializeSecureRandomRelatedClassesAtRuntime(
     @BuildStep
     void createNatsConnector(BuildProducer<AdditionalBeanBuildItem> buildProducer) {
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(JetStreamConnector.class));
-        buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(JetStreamInstrumenter.class));
+        buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(JetStreamInstrument.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(ExecutionHolder.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(ConnectionFactory.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(DefaultPayloadMapper.class));

deployment/src/test/java/io/quarkiverse/reactive/messaging/nats/jetstream/test/DeadLetterConsumingBean.java

@@ -17,6 +17,7 @@
 import io.quarkiverse.reactive.messaging.nats.NatsConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.Connection;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
+import io.quarkiverse.reactive.messaging.nats.jetstream.client.DefaultConnectionListener;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConnectionConfiguration;
 import io.smallrye.mutiny.Uni;
 
@@ -83,8 +84,8 @@ private Uni<Connection> getOrEstablishConnection() {
                 .filter(Connection::isConnected)
                 .orElse(null))
                 .onItem().ifNull()
-                .switchTo(() -> connectionFactory.create(ConnectionConfiguration.of(natsConfiguration), (event, message) -> {
-                }))
+                .switchTo(() -> connectionFactory.create(ConnectionConfiguration.of(natsConfiguration),
+                        new DefaultConnectionListener()))
                 .onItem().invoke(this.connection::set);
     }
 }

deployment/src/test/java/io/quarkiverse/reactive/messaging/nats/jetstream/test/ExponentialBackoffConsumingBean.java

@@ -17,6 +17,7 @@
 import io.quarkiverse.reactive.messaging.nats.NatsConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.Connection;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
+import io.quarkiverse.reactive.messaging.nats.jetstream.client.DefaultConnectionListener;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConnectionConfiguration;
 import io.smallrye.mutiny.Uni;
 
@@ -93,8 +94,7 @@ private Uni<Connection> getOrEstablishConnection() {
                 .filter(Connection::isConnected)
                 .orElse(null))
                 .onItem().ifNull().switchTo(() -> connectionFactory
-                        .create(ConnectionConfiguration.of(natsConfiguration), (event, message) -> {
-                        }))
+                        .create(ConnectionConfiguration.of(natsConfiguration), new DefaultConnectionListener()))
                 .onItem().invoke(this.connection::set);
     }
 }

deployment/src/test/java/io/quarkiverse/reactive/messaging/nats/jetstream/test/KeyValueStoreResource.java

@@ -15,6 +15,7 @@
 import io.quarkiverse.reactive.messaging.nats.NatsConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.Connection;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
+import io.quarkiverse.reactive.messaging.nats.jetstream.client.DefaultConnectionListener;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConnectionConfiguration;
 import io.smallrye.mutiny.Uni;
 
@@ -84,8 +85,8 @@ private Uni<Connection> getOrEstablishConnection() {
                 .filter(Connection::isConnected)
                 .orElse(null))
                 .onItem().ifNull()
-                .switchTo(() -> connectionFactory.create(ConnectionConfiguration.of(natsConfiguration), (event, message) -> {
-                }))
+                .switchTo(() -> connectionFactory.create(ConnectionConfiguration.of(natsConfiguration),
+                        new DefaultConnectionListener()))
                 .onItem().invoke(this.connection::set);
     }
 }

deployment/src/test/java/io/quarkiverse/reactive/messaging/nats/jetstream/test/ReaderSubscribeConnectionTest.java

@@ -19,13 +19,14 @@
 import io.nats.client.api.ReplayPolicy;
 import io.quarkiverse.reactive.messaging.nats.NatsConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
+import io.quarkiverse.reactive.messaging.nats.jetstream.client.DefaultConnectionListener;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConnectionConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConsumerConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ReaderConsumerConfiguration;
 import io.quarkus.test.QuarkusUnitTest;
 
 public class ReaderSubscribeConnectionTest {
-    private Logger logger = Logger.getLogger(ReaderSubscribeConnectionTest.class);
+    private final static Logger logger = Logger.getLogger(ReaderSubscribeConnectionTest.class);
 
     @RegisterExtension
     static QuarkusUnitTest runner = new QuarkusUnitTest()
@@ -42,30 +43,30 @@ void createConnectionWithModifiedConfiguration() throws Exception {
         var consumerConfiguration = createConsumerConfiguration(List.of(Duration.ofSeconds(10)), 2L);
 
         try (final var connection = connectionFactory.create(ConnectionConfiguration.of(natsConfiguration),
-                (event, message) -> {
-                },
-                consumerConfiguration).await().atMost(Duration.ofSeconds(30))) {
+                new DefaultConnectionListener()).await().atMost(Duration.ofSeconds(30))) {
             logger.info("Connected to NATS");
-            final var consumer = connection.getConsumer("reader-test", consumerConfiguration.consumerConfiguration().name())
-                    .await().atMost(Duration.ofSeconds(30));
-            assertThat(consumer).isNotNull();
-            assertThat(consumer.configuration().backoff()).isEqualTo(List.of(Duration.ofSeconds(10)));
-            assertThat(consumer.configuration().maxDeliver()).isEqualTo(2L);
+            try (final var ignored = connection.subscribtion(consumerConfiguration).await().atMost(Duration.ofSeconds(30))) {
+                final var consumer = connection.getConsumer("reader-test", consumerConfiguration.consumerConfiguration().name())
+                        .await().atMost(Duration.ofSeconds(30));
+                assertThat(consumer).isNotNull();
+                assertThat(consumer.configuration().backoff()).isEqualTo(List.of(Duration.ofSeconds(10)));
+                assertThat(consumer.configuration().maxDeliver()).isEqualTo(2L);
+            }
         }
 
         consumerConfiguration = createConsumerConfiguration(List.of(Duration.ofSeconds(10), Duration.ofSeconds(30)), 3L);
         try (final var connection = connectionFactory.create(ConnectionConfiguration.of(natsConfiguration),
-                (event, message) -> {
-                },
-                consumerConfiguration).await().atMost(Duration.ofSeconds(30))) {
-            logger.info("Connected to NATS");
-            final var consumer = connection.getConsumer("reader-test", consumerConfiguration.consumerConfiguration().name())
-                    .await().atMost(Duration.ofSeconds(30));
-            assertThat(consumer).isNotNull();
-            assertThat(consumer.configuration().backoff()).isEqualTo(List.of(Duration.ofSeconds(10), Duration.ofSeconds(30)));
-            assertThat(consumer.configuration().maxDeliver()).isEqualTo(3L);
+                new DefaultConnectionListener()).await().atMost(Duration.ofSeconds(30))) {
+            try (final var ignored = connection.subscribtion(consumerConfiguration).await().atMost(Duration.ofSeconds(30))) {
+                logger.info("Connected to NATS");
+                final var consumer = connection.getConsumer("reader-test", consumerConfiguration.consumerConfiguration().name())
+                        .await().atMost(Duration.ofSeconds(30));
+                assertThat(consumer).isNotNull();
+                assertThat(consumer.configuration().backoff())
+                        .isEqualTo(List.of(Duration.ofSeconds(10), Duration.ofSeconds(30)));
+                assertThat(consumer.configuration().maxDeliver()).isEqualTo(3L);
+            }
         }
-
     }
 
     private ReaderConsumerConfiguration<Object> createConsumerConfiguration(List<Duration> backoff, Long maxDeliver) {

deployment/src/test/java/io/quarkiverse/reactive/messaging/nats/jetstream/test/RequestReplyResource.java

@@ -27,6 +27,7 @@
 import io.quarkiverse.reactive.messaging.nats.jetstream.JetStreamOutgoingMessageMetadata;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.Connection;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
+import io.quarkiverse.reactive.messaging.nats.jetstream.client.DefaultConnectionListener;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.api.StreamState;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConnectionConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.FetchConsumerConfiguration;
@@ -108,8 +109,8 @@ private Uni<Connection> getOrEstablishMessageConnection() {
                 .filter(Connection::isConnected)
                 .orElse(null))
                 .onItem().ifNull()
-                .switchTo(() -> connectionFactory.create(ConnectionConfiguration.of(natsConfiguration), (event, message) -> {
-                }))
+                .switchTo(() -> connectionFactory.create(ConnectionConfiguration.of(natsConfiguration),
+                        new DefaultConnectionListener()))
                 .onItem().invoke(this.messageConnection::set);
     }
 

docs/modules/ROOT/pages/includes/attributes.adoc

@@ -1,3 +1,3 @@
-:project-version: 3.15.5
+:project-version: 3.15.6
 
 :examples-dir: ./../examples/

runtime/pom.xml

@@ -29,6 +29,10 @@
             <groupId>io.opentelemetry.instrumentation</groupId>
             <artifactId>opentelemetry-instrumentation-api-semconv</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.nats</groupId>
             <artifactId>jnats</artifactId>

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/JetStreamConnector.java

@@ -17,6 +17,7 @@
 import org.eclipse.microprofile.config.Config;
 import org.eclipse.microprofile.reactive.messaging.Message;
 import org.eclipse.microprofile.reactive.messaging.spi.Connector;
+import org.jboss.logging.Logger;
 
 import io.quarkiverse.reactive.messaging.nats.NatsConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
@@ -77,6 +78,8 @@
 public class JetStreamConnector implements InboundConnector, OutboundConnector, HealthReporter {
     public static final String CONNECTOR_NAME = "quarkus-jetstream";
 
+    private final static Logger logger = Logger.getLogger(JetStreamConnector.class);
+
     private final List<MessageProcessor> processors;
     private final NatsConfiguration natsConfiguration;
     private final ConnectionFactory connectionFactory;
@@ -132,7 +135,14 @@ public HealthReport getLiveness() {
 
     public void terminate(
             @Observes(notifyObserver = Reception.IF_EXISTS) @Priority(50) @BeforeDestroyed(ApplicationScoped.class) Object ignored) {
-        this.processors.forEach(processor -> processor.close().await().indefinitely());
+        this.processors.forEach(processor -> {
+            try {
+                processor.close();
+            } catch (Throwable failure) {
+                logger.warnf(failure, "Failed to close the processor: %s", failure.getMessage());
+            }
+        });
+        this.processors.clear();
     }
 
     private MessagePublisherProcessor<?> createMessagePublisherProcessor(

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/Connection.java

@@ -8,8 +8,7 @@
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.api.Consumer;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.api.PurgeResult;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.api.StreamState;
-import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.FetchConsumerConfiguration;
-import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.PublishConfiguration;
+import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.*;
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
 
@@ -23,6 +22,8 @@ public interface Connection extends AutoCloseable {
 
     void addListener(ConnectionListener listener);
 
+    void removeListener(ConnectionListener listener);
+
     default void fireEvent(ConnectionEvent event, String message) {
         listeners().forEach(listener -> listener.onEvent(event, message));
     }
@@ -70,4 +71,10 @@ <T> Uni<Message<T>> publish(final Message<T> message, final PublishConfiguration
     Uni<Void> deleteKeyValue(String bucketName, String key);
 
     <T> Uni<Message<T>> resolve(String streamName, long sequence);
+
+    <T> Uni<Subscription<T>> subscribtion(PushConsumerConfiguration<T> configuration);
+
+    <T> Uni<Subscription<T>> subscribtion(ReaderConsumerConfiguration<T> configuration);
+
+    <T> void close(Subscription<T> subscribtion);
 }

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/ConnectionEvent.java

@@ -5,6 +5,5 @@ public enum ConnectionEvent {
     Connected,
     Closed,
     Reconnected,
-    CommunicationFailed,
-    SubscriptionInactive,
+    CommunicationFailed
 }

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/ConnectionFactory.java

@@ -9,13 +9,11 @@
 import org.jboss.logging.Logger;
 
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ConnectionConfiguration;
-import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.PushConsumerConfiguration;
-import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.ReaderConsumerConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.ConsumerMapper;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.MessageMapper;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.PayloadMapper;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.StreamStateMapper;
-import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrumenter;
+import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrument;
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.unchecked.Unchecked;
 import io.smallrye.reactive.messaging.providers.connectors.ExecutionHolder;
@@ -31,15 +29,15 @@ public class ConnectionFactory {
 
     private final ExecutionHolder executionHolder;
     private final MessageMapper messageMapper;
-    private final JetStreamInstrumenter instrumenter;
+    private final JetStreamInstrument instrumenter;
     private final PayloadMapper payloadMapper;
     private final ConsumerMapper consumerMapper;
     private final StreamStateMapper streamStateMapper;
 
     @Inject
     public ConnectionFactory(ExecutionHolder executionHolder,
             MessageMapper messageMapper,
-            JetStreamInstrumenter instrumenter,
+            JetStreamInstrument instrumenter,
             PayloadMapper payloadMapper,
             ConsumerMapper consumerMapper,
             StreamStateMapper streamStateMapper) {
@@ -51,41 +49,6 @@ public ConnectionFactory(ExecutionHolder executionHolder,
         this.streamStateMapper = streamStateMapper;
     }
 
-    public <T> Uni<? extends SubscribeConnection<T>> create(ConnectionConfiguration connectionConfiguration,
-            ConnectionListener connectionListener,
-            ReaderConsumerConfiguration<T> consumerConfiguration) {
-        return getContext()
-                .onItem()
-                .transformToUni(context -> Uni.createFrom()
-                        .item(Unchecked.supplier(() -> new DefaultConnection(connectionConfiguration, connectionListener,
-                                context, messageMapper, payloadMapper, consumerMapper, streamStateMapper, instrumenter))))
-                .onItem().transformToUni(connection -> Uni.createFrom()
-                        .item(Unchecked.supplier(() -> new ReaderSubscribeConnection<>(connection, consumerConfiguration))))
-                .onFailure().invoke(failure -> logger.errorf(failure, "Failed connecting to NATS: %s", failure.getMessage()))
-                .onFailure()
-                .retry()
-                .withBackOff(connectionConfiguration.connectionBackoff().orElse(DEFAULT_CONNECTION_BACKOFF))
-                .atMost(connectionConfiguration.connectionAttempts().orElse(DEFAULT_CONNECTION_ATTEMPTS));
-    }
-
-    public <T> Uni<? extends SubscribeConnection<T>> create(ConnectionConfiguration connectionConfiguration,
-            ConnectionListener connectionListener,
-            PushConsumerConfiguration<T> consumerConfiguration) {
-
-        return getContext()
-                .onItem()
-                .transformToUni(context -> Uni.createFrom()
-                        .item(Unchecked.supplier(() -> new DefaultConnection(connectionConfiguration, connectionListener,
-                                context, messageMapper, payloadMapper, consumerMapper, streamStateMapper, instrumenter))))
-                .onItem().transformToUni(connection -> Uni.createFrom()
-                        .item(Unchecked.supplier(() -> new PushSubscribeConnection<>(connection, consumerConfiguration))))
-                .onFailure().invoke(failure -> logger.errorf(failure, "Failed connecting to NATS: %s", failure.getMessage()))
-                .onFailure()
-                .retry()
-                .withBackOff(connectionConfiguration.connectionBackoff().orElse(DEFAULT_CONNECTION_BACKOFF))
-                .atMost(connectionConfiguration.connectionAttempts().orElse(DEFAULT_CONNECTION_ATTEMPTS));
-    }
-
     public Uni<? extends Connection> create(ConnectionConfiguration connectionConfiguration,
             ConnectionListener connectionListener) {
         return getContext()

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/ConnectionListener.java

@@ -1,6 +1,6 @@
 package io.quarkiverse.reactive.messaging.nats.jetstream.client;
 
-public interface ConnectionListener {
+public interface ConnectionListener extends AutoCloseable {
 
     void onEvent(ConnectionEvent event, String message);