Repository: quarkusio/quarkus
Issue: Quarkus redis reactive pubsub disconnecting handler

### Describe the bug

I have the following code:
```
ReactiveRedisDataSource redis; -> Injected
ReactivePubSubCommands<String> pubsub = redis.pubsub(String.class);
pubsub.subscribe('my-channel')
                .onItem().call(this::consume)
                .subscribe().with(
                        notificationId -> log.debug("New notification has been processed {}", notificationId),
                        err -> log.error("Error while listening for new notifications ", err));
```

Everything is working fine, I receive data via the subscription, but for some reason, after a while (days, weeks), my subscription is gone. It can be checked by running in redis the folowing command:
```
PUBSUB NUMSUB my-channel
```
Initially returns 1, but when my issues occurs, it is going to 0. No other failures can be seen in the application. 

It is very easy to reproduce the issue:
1. Connect via the code above to a redis docker instance.
2. Check `PUBSUB NUMSUB my-channel -> 1`
3. Stop redis container
4. Start redis container
5. Now `PUBSUB NUMSUB my-channel -> 0` and no errors thrown in code.


**How can I watch the disconnected state to be able to manually reconnect?** I tried to add handlers like `onTermination`, `onCanncelation`, `onFailure`, `onCompletion` but they are never triggered. 


An undesired solution is to use 
```
pubsub.Uni<ReactiveRedisSubscriber> subscribe(String channel, Consumer<V> onMessage, Runnable onEnd,
            Consumer<Throwable> onException);
```

I noticed that when my redis connection is going down, it triggers `onEnd` handler.

But this cuts my reactivity in the process for consuming the event. I want to use `pubsub.Multi<V> subscribe(String... channels)`


### Quarkus version or git rev

2.14.3.Final


Changed files:
extensions/redis-client/runtime/src/main/java/io/quarkus/redis/runtime/datasource/ReactivePubSubCommandsImpl.java

diff --git a/extensions/redis-client/runtime/src/main/java/io/quarkus/redis/runtime/datasource/ReactivePubSubCommandsImpl.java b/extensions/redis-client/runtime/src/main/java/io/quarkus/redis/runtime/datasource/ReactivePubSubCommandsImpl.java
index c06075075d3..95476dcb225 100644
--- a/extensions/redis-client/runtime/src/main/java/io/quarkus/redis/runtime/datasource/ReactivePubSubCommandsImpl.java
+++ b/extensions/redis-client/runtime/src/main/java/io/quarkus/redis/runtime/datasource/ReactivePubSubCommandsImpl.java
@@ -143,9 +143,11 @@ public Multi<V> subscribe(String... channels) {
         doesNotContainNull(channels, "channels");
 
         return Multi.createFrom().emitter(emitter -> {
-            subscribe(List.of(channels), emitter::emit)
+            subscribe(List.of(channels), emitter::emit, emitter::complete, emitter::fail)
                     .subscribe().with(x -> {
-                        emitter.onTermination(() -> x.unsubscribe(channels).subscribe().asCompletionStage());
+                        emitter.onTermination(() -> {
+                            x.unsubscribe(channels).subscribe().asCompletionStage();
+                        });
                     }, emitter::fail);
         });
     }
@@ -156,7 +158,7 @@ public Multi<V> subscribeToPatterns(String... patterns) {
         doesNotContainNull(patterns, "patterns");
 
         return Multi.createFrom().emitter(emitter -> {
-            subscribeToPatterns(List.of(patterns), emitter::emit)
+            subscribeToPatterns(List.of(patterns), emitter::emit, emitter::complete, emitter::fail)
                     .subscribe().with(x -> {
                         emitter.onTermination(() -> x.unsubscribe(patterns).subscribe().asCompletionStage());
                     }, emitter::fail);
@@ -186,10 +188,12 @@ private AbstractRedisSubscriber(RedisConnection connection, RedisAPI api, Consum
         public Uni<String> subscribe() {
             Uni<Void> handled = Uni.createFrom().emitter(emitter -> {
                 connection.handler(r -> runOnDuplicatedContext(() -> handleRedisEvent(emitter, r)));
-                if (onEnd != null)
+                if (onEnd != null) {
                     connection.endHandler(() -> runOnDuplicatedContext(onEnd));
-                if (onException != null)
+                }
+                if (onException != null) {
                     connection.exceptionHandler(t -> runOnDuplicatedContext(() -> onException.accept(t)));
+                }
             });
 
             Uni<Void> subscribed = subscribeToRedis();