Added nextMessages method to connection (#185)

* Added nextMessages method to connection --------- Co-authored-by: Paw Figgé Kjeldgaard <[email protected]>
quarkiverse · Sep 9, 2024 · e3054db · e3054db
1 parent 5a7aedb
commit e3054db
Show file tree

Hide file tree

Showing 4 changed files with 41 additions and 2 deletions.
diff --git a/.github/project.yml b/.github/project.yml
@@ -1,4 +1,4 @@
 release:
-  current-version: "1.13.0"
+  current-version: "1.13.1"
   next-version: "1.14.0-SNAPSHOT"
 
diff --git a/docs/modules/ROOT/pages/includes/attributes.adoc b/docs/modules/ROOT/pages/includes/attributes.adoc
@@ -1,3 +1,3 @@
-:project-version: 1.13.0
+:project-version: 1.13.1
 
 :examples-dir: ./../examples/
diff --git a/.../main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/MessageConnection.java b/.../main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/MessageConnection.java
@@ -4,6 +4,7 @@
 
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.FetchConsumerConfiguration;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.configuration.PublishConfiguration;
+import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
 
 public interface MessageConnection extends Connection {
@@ -15,6 +16,8 @@ <T> Uni<Message<T>> publish(final Message<T> message, final PublishConfiguration
 
     <T> Uni<Message<T>> nextMessage(FetchConsumerConfiguration<T> configuration);
 
+    <T> Multi<Message<T>> nextMessages(FetchConsumerConfiguration<T> configuration);
+
     <T> Uni<T> getKeyValue(String bucketName, String key, Class<T> valueType);
 
     <T> Uni<Void> putKeyValue(String bucketName, String key, T value);

diff --git a/...java/io/quarkiverse/reactive/messaging/nats/jetstream/client/vertx/MessageConnection.java b/...java/io/quarkiverse/reactive/messaging/nats/jetstream/client/vertx/MessageConnection.java
@@ -26,6 +26,7 @@
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.message.MessageFactory;
 import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrumenter;
 import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamTrace;
+import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.unchecked.Unchecked;
 import io.vertx.mutiny.core.Context;
@@ -116,6 +117,14 @@ public <T> Uni<Message<T>> nextMessage(FetchConsumerConfiguration<T> configurati
                 .transformToUni(consumerContext -> nextMessage(consumerContext, configuration));
     }
 
+    @Override
+    public <T> Multi<Message<T>> nextMessages(FetchConsumerConfiguration<T> configuration) {
+        return getConsumerContext(connection, context, configuration.stream(),
+                configuration.name().orElseThrow(() -> new IllegalArgumentException("Consumer name is not configured")))
+                .onItem().transformToMulti(consumerContext -> nextMessages(consumerContext, configuration))
+                .emitOn(context::runOnContext);
+    }
+
     @Override
     public <T> Uni<T> getKeyValue(String bucketName, String key, Class<T> valueType) {
         return Uni.createFrom().item(Unchecked.supplier(() -> {
@@ -268,4 +277,31 @@ private <T> Uni<Void> addOrUpdateConsumer(FetchConsumerConfiguration<T> configur
         }))
                 .emitOn(context::runOnContext);
     }
+
+    private <T> Multi<Message<T>> nextMessages(final ConsumerContext consumerContext,
+            FetchConsumerConfiguration<T> configuration) {
+        ExecutorService executor = Executors.newSingleThreadExecutor(JetstreamWorkerThread::new);
+        return Multi.createFrom().<Message<T>> emitter(emitter -> {
+            try {
+                try (final var fetchConsumer = fetchConsumer(consumerContext, configuration.fetchTimeout().orElse(null))) {
+                    var message = fetchConsumer.nextMessage();
+                    while (message != null) {
+                        emitter.emit(messageFactory.create(
+                                message,
+                                configuration.traceEnabled(),
+                                configuration.payloadType().orElse(null),
+                                context,
+                                new ExponentialBackoff(false, Duration.ZERO),
+                                configuration.ackTimeout()));
+                        message = fetchConsumer.nextMessage();
+                    }
+                    emitter.complete();
+                }
+            } catch (Throwable failure) {
+                emitter.fail(new FetchException(failure));
+            }
+        })
+                .emitOn(context::runOnContext)
+                .runSubscriptionOn(executor);
+    }
 }