#27 documentation and broadcast consuming

Author: petrenyi-mark

docs/connector/configuration-reference.adoc

@@ -47,11 +47,17 @@ For example if you use `quarkus.redis.*my-redis*.hosts` configuration then you c
 | INCOMING
 
 | `xread-noack`
-| Include the `NOACK` parameter in the `XREADGROUP` call
+| Include the `NOACK` parameter in the `XREADGROUP` call.
 | `true`
 | No
 | INCOMING
 
+| `broadcast`
+| Allow the received entries to be consumed by multiple channels.
+| `false`
+| No
+| INCOMING
+
 | `xadd-maxlen`
 | The maximum number of entries to keep in the stream (trims old entries).
 | -

docs/migration/migration110to120.adoc

@@ -3,7 +3,17 @@
 == reactive-messaging-redisstream-connector
 
 * RedisStreamsConnector#prudentRun handling fix, separate handling of the flag per config.
+* Consume messages on vert.x context to retain context data in all cases.
+* Added `broadcast` incoming configuration to allow multiple consumers on the same channel (ie. multiple methods with `@Incoming` for the same channel).
 
 === Migration
 
 Changes are backwards compatible doesn't need any migration.
+
+== quarkus-reactive-messaging-redisstream-extension-sample
+
+* Added both imperative and reactive example for `@Incoming` usage.
+
+=== Migration
+
+Changes are backwards compatible doesn't need any migration.

quarkus-reactive-messaging-redisstream-extension-sample/src/main/java/hu/icellmobilsoft/quarkus/sample/MyMessagingApplication.java

@@ -19,14 +19,16 @@
  */
 package hu.icellmobilsoft.quarkus.sample;
 
-import hu.icellmobilsoft.reactive.messaging.redis.streams.metadata.IncomingRedisStreamMetadata;
-import io.quarkus.logging.Log;
-import io.quarkus.runtime.StartupEvent;
-import io.smallrye.mutiny.Uni;
-import io.smallrye.reactive.messaging.annotations.Blocking;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.event.Observes;
 import jakarta.inject.Inject;
+
 import org.eclipse.microprofile.faulttolerance.Retry;
 import org.eclipse.microprofile.reactive.messaging.Channel;
 import org.eclipse.microprofile.reactive.messaging.Emitter;
@@ -35,10 +37,11 @@
 import org.eclipse.microprofile.reactive.messaging.Outgoing;
 import org.jboss.logging.MDC;
 
-import java.util.Optional;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
+import hu.icellmobilsoft.reactive.messaging.redis.streams.metadata.IncomingRedisStreamMetadata;
+import io.quarkus.logging.Log;
+import io.quarkus.runtime.StartupEvent;
+import io.smallrye.mutiny.Uni;
+import io.smallrye.reactive.messaging.annotations.Blocking;
 
 /**
  * Class that handles messages
@@ -99,6 +102,32 @@ public void sendMessage(String message) {
     @Blocking(ordered = false, value = "incoming-pool")
     @Retry(maxRetries = 2)
     public String toUpperCase(String payload) {
+        return doAction(payload);
+    }
+
+    /**
+     * Consume the message from the "words-in" channel using reactive signature.
+     * 
+     * @param message
+     *            reactive message containing the payload
+     *
+     * @return a void completion stage
+     */
+    @Incoming("words-in")
+    @Blocking(ordered = false, value = "incoming-pool")
+    @Retry(maxRetries = 2)
+    public CompletionStage<Void> processReactive(Message<String> message) {
+        try {
+            String result = doAction(message.getPayload());
+            Log.infov("Processed message: {0}", result);
+            return message.ack();
+        } catch (Exception e) {
+            Log.error("Error during logging metadata", e);
+            return message.nack(e);
+        }
+    }
+
+    private String doAction(String payload) {
         Log.infov("Message received: [{0}]", payload);
         try {
             TimeUnit.MILLISECONDS.sleep(2000);

quarkus-reactive-messaging-redisstream-extension-sample/src/main/resources/application.yml

@@ -10,6 +10,7 @@ mp:
         stream-key: sample-stream
         xread-block-ms: 60000
         #connection-key: redis2
+        broadcast: true
     outgoing:
       words-out:
         connector: reactive-messaging-redis-streams

reactive-messaging-redisstream-connector/src/main/java/hu/icellmobilsoft/reactive/messaging/redis/streams/RedisStreamsConnector.java

@@ -90,6 +90,8 @@
 @ConnectorAttribute(name = "xread-noack", description = "Include the NOACK parameter in the XREADGROUP call", type = "boolean",
         defaultValue = "true",
         direction = ConnectorAttribute.Direction.INCOMING)
+@ConnectorAttribute(name = "broadcast", description = "Allow the received entries to be consumed by multiple channels", type = "boolean", defaultValue = "false",
+        direction = ConnectorAttribute.Direction.INCOMING)
 @ConnectorAttribute(name = "xadd-maxlen", description = "The maximum number of entries to keep in the stream", type = "int",
         direction = ConnectorAttribute.Direction.OUTGOING)
 @ConnectorAttribute(name = "xadd-exact-maxlen", description = "Use exact trimming for MAXLEN parameter", type = "boolean", defaultValue = "false",
@@ -205,7 +207,11 @@ public Flow.Publisher<? extends Message<?>> getPublisher(Config config) {
 
         RedisStreams redisAPI = redisStreamsProducer.produce(incomingConfig.getConnectionKey());
         redisStreams.add(redisAPI);
-        return xreadMulti(redisAPI, incomingConfig);
+        Multi<Message<Object>> publisher = xreadMulti(redisAPI, incomingConfig);
+        if(Boolean.TRUE.equals(incomingConfig.getBroadcast())){
+            publisher = publisher.broadcast().toAllSubscribers();
+        }
+        return publisher;
     }
 
     /**