Merge pull request quarkusio#50124 from Ladicek/redis-improvements

Redis: improvements

Author: cescoffier

docs/src/main/asciidoc/redis-reference.adoc

@@ -643,7 +643,7 @@ There are other methods that can be useful to manipulate counters, such as:
 - `set` - to set an initial value if needed
 - `decr` and `decrby` - allows decrementing the stored value
 
-==== Communicate with pub/sub
+=== Communicate with pub/sub
 
 Redis allows sending _messages_ to channels and listening for these messages.
 These features are available from the `pubsub` group.
@@ -710,7 +710,7 @@ public static class MyCache {
 }
 ----
 
-==== Use Redis transactions
+=== Use Redis transactions
 
 Redis transactions are slightly different from relational database transactions.
 Redis transactions are a batch of commands executed altogether.
@@ -741,7 +741,7 @@ TransactionResult result = ds.withTransaction(tx -> {
     });
 ----
 
-The received `tx` object can also be used to _discard_ the transaction, using: `tx.discard();`.
+The received `tx` object can also be used to _discard_ the transaction, using `tx.discard()`.
 The returned `TransactionResult` lets you retrieve the result of each command.
 
 When using the reactive variant of the data source, the passed callback is a `Function<ReactiveTransactionalRedisDataSource, Uni<Void>>`:
@@ -761,7 +761,7 @@ Uni<TransactionResult> result = ds.withTransaction(tx -> {
 
 Transaction execution can be conditioned by _keys_.
 When a passed key gets modified during the execution of a transaction, the transaction is discarded.
-The keys are passed as `String` as a second parameter to the `withTransaction` method:
+The keys are passed as ``String``s to the second, vararg parameter of the `withTransaction` method:
 
 [source,java]
 ----
@@ -776,19 +776,21 @@ IMPORTANT: You cannot use the pub/sub feature from within a transaction.
 
 ==== Implement the optimistic locking pattern
 
-To use optimistic locking, you need to use a variant of the `withTransaction` method, allowing the execution of code before the transaction starts.
+To use optimistic locking, you need to use a variant of the `withTransaction` method allowing execution of code before the transaction starts.
 In other words, it will be executed as follows:
 
 [source]
 ----
 WATCH key
 
 // Pre-transaction block
-// ....
+// ...
 // Produce a result
 
 MULTI
-  // In transaction code, receive the result produced by the pre-transaction block.
+
+// In transaction code, receive the result produced by the pre-transaction block.
+
 EXEC
 ----
 
@@ -801,14 +803,16 @@ OptimisticLockingTransactionResult<Boolean> result = blocking.withTransaction(ds
     HashCommands<String, String, String> hashCommands = ds.hash(String.class);
     return hashCommands.hexists(key, "field"); // Produce a result (boolean in this case)
 },
- (exists, tx) -> { // The transactional block, receives the result and the transactional data source
-        if (exists) {
-            tx.hash(String.class).hset(key, "field", "new value");
-        } else {
-            tx.discard();
-        }
- },
-  key); // The watched key
+(exists, tx) -> {
+    // The transactional block, receives the result and the transactional data source
+    if (exists) {
+        tx.hash(String.class).hset(key, "field", "new value");
+    } else {
+        tx.discard();
+    }
+},
+// The watched key
+key);
 ----
 
 If one of the watched keys is touched before or during the execution of the pre-transaction or transactional blocks, the transaction is aborted.
@@ -821,9 +825,35 @@ Consequently, the pre-transaction block must use the passed data source to execu
 Thus, the commands are emitted from that connection.
 These commands must not modify the watched keys.
 
-The transaction is aborted if the pre-transaction block throws an exception (or produces a failure when using the reactive API).
+==== Error handling
+
+If the transaction block returns successfully (or completes with an item in case of the reactive API), the transaction is executed.
+Note that in this case, some transaction commands may succeed and some may fail; this is standard Redis behavior.
+The `TransactionResult.hasErrors()` method indicates whether at least one command failed.
+Results of failed commands in the `TransactionResult` are represented as ``Throwable``s.
+If the transaction block throws an exception (or completes with a failure in case of the reactive API), the transaction is discarded automatically and the exception is rethrown (or the resulting `Uni` completes with the same failure in case of the reactive API).
+
+In case of the optimistic locking API, if the pre-transaction block returns successfully (or completes with an item in case of the reactive API), the transaction is started and execution proceeds to the transaction block.
+If the pre-transaction block throws an exception (or completes with a failure in case of the reactive API), all watched keys are ``UNWATCH``ed automatically and the exception is rethrown (or the resulting `Uni` completes with the same failure in case of the reactive API).
+In this case, the transaction is not started and the transaction block is not executed.
+
+==== Transactions in Redis Cluster
+
+In the cluster mode, transaction commands (`MULTI`, `EXEC`, `DISCARD`, `WATCH`, `UNWATCH`) are handled specially.
+
+By default, transactions in cluster are disabled and transaction commands immediately fail.
+
+It is however possible to enable single-node transactions by setting `quarkus.redis.cluster-transactions` to `single-node`.
+Note that this setting only makes sense in the cluster mode and is ignored otherwise.
+
+In this mode, the `MULTI` command is queued and is only issued when the next command is executed.
+This next command binds the connection to the corresponding node of the Redis cluster (so it should have keys, otherwise the target node is random).
+All subsequent commands are targeted to that node.
+If some of the subsequent commands have a key that belongs to another node, the command fails on the server side.
+If `WATCH` is used before `MULTI`, its key(s) determine to which node the connection is bound and the subsequent `MULTI` is not queued.
+If `WATCH` keys belong to multiple nodes, the command fails on the client side.
 
-==== Execute custom commands
+=== Execute custom commands
 
 To execute a custom command, or a command not supported by the API, use the following approach:
 
@@ -927,8 +957,8 @@ The documentation of the Vert.x Redis Client is available on the https://vertx.i
 
 == Configure Redis hosts programmatically
 
-The `RedisHostsProvider` programmatically provides redis hosts.
-This allows for configuration of properties like redis connection password coming from other sources.
+The `RedisHostsProvider` programmatically provides Redis hosts.
+This allows for configuration of properties like Redis connection password coming from other sources.
 
 [NOTE]
 ====

extensions/redis-client/runtime/src/main/java/io/quarkus/redis/datasource/ReactiveRedisDataSource.java

@@ -30,7 +30,6 @@
 import io.quarkus.redis.datasource.transactions.OptimisticLockingTransactionResult;
 import io.quarkus.redis.datasource.transactions.ReactiveTransactionalRedisDataSource;
 import io.quarkus.redis.datasource.transactions.TransactionResult;
-import io.quarkus.redis.datasource.transactions.TransactionalRedisDataSource;
 import io.quarkus.redis.datasource.value.ReactiveValueCommands;
 import io.smallrye.common.annotation.Experimental;
 import io.smallrye.mutiny.Uni;
@@ -50,84 +49,99 @@
 public interface ReactiveRedisDataSource {
 
     /**
-     * Retrieves a {@link ReactiveRedisDataSource} using a single connection with the Redis Server.
-     * The connection is acquired from the pool and released when the {@code Uni} returned by {@code function} produces
-     * a {@code null} item or a failure.
+     * Obtains a {@link ReactiveRedisDataSource} that uses a single connection to the Redis server
+     * and passes it to the given {@code function}. The connection is acquired from the pool
+     * and released when the {@code Uni} returned by {@code function} produces an item or a failure.
      *
-     * @param function the function receiving the single-connection data source and producing {@code null} when the
+     * @param function the function receiving the connection and producing {@code null} when the
      *        connection can be released.
      */
     Uni<Void> withConnection(Function<ReactiveRedisDataSource, Uni<Void>> function);
 
     /**
-     * Retrieves a {@link RedisDataSource} enqueuing commands in a Redis Transaction ({@code MULTI}).
-     * Note that transaction acquires a single connection, and all the commands are enqueued in this connection.
-     * The commands are only executed when the passed block emits the {@code null} item.
+     * Obtains a {@link ReactiveRedisDataSource} that enqueues commands in a Redis Transaction ({@code MULTI})
+     * and passes it to the given {@code tx} block. Note that the transaction acquires a single connection
+     * and all the commands are enqueued on this connection. The commands are only executed when the {@code Uni}
+     * returned by {@code tx} produces an item.
      * <p>
-     * The results of the commands are retrieved using the produced {@link TransactionResult}.
+     * The results of the commands can be obtained from the returned {@link TransactionResult}.
+     * Errors are represented as {@code Throwable}s in the {@code TransactionResult}.
      * <p>
-     * The user can discard a transaction using the {@link TransactionalRedisDataSource#discard()} method.
-     * In this case, the produced {@link TransactionResult} will be empty.
+     * The user can discard a transaction using the {@link ReactiveTransactionalRedisDataSource#discard()} method.
+     * In this case, the produced {@link TransactionResult} is empty.
+     * If the {@code tx} block completes with a failure, the transaction is discarded automatically and
+     * the resulting {@code Uni} completes with the same failure.
      *
-     * @param tx the consumer receiving the transactional redis data source. The enqueued commands are only executed
-     *        at the end of the block.
+     * @param tx the consumer receiving the transactional Redis data source. The enqueued commands are only executed
+     *        when this block completes with an item.
      */
     Uni<TransactionResult> withTransaction(Function<ReactiveTransactionalRedisDataSource, Uni<Void>> tx);
 
     /**
-     * Retrieves a {@link RedisDataSource} enqueuing commands in a Redis Transaction ({@code MULTI}).
-     * Note that transaction acquires a single connection, and all the commands are enqueued in this connection.
-     * The commands are only executed when the passed block emits the {@code null} item.
+     * Obtains a {@link ReactiveRedisDataSource} that enqueues commands in a Redis Transaction ({@code WATCH} & {@code MULTI}),
+     * starts watching the given {@code watchedKeys}, and passes it to the given {@code tx} block.
+     * Note that the transaction acquires a single connection and all the commands are enqueued on this connection.
+     * The commands are only executed when the {@code Uni} returned by {@code tx} produces an item.
      * <p>
-     * The results of the commands are retrieved using the produced {@link TransactionResult}.
+     * The results of the commands can be obtained from the returned {@link TransactionResult}.
+     * Errors are represented as {@code Throwable}s in the {@code TransactionResult}.
      * <p>
-     * The user can discard a transaction using the {@link TransactionalRedisDataSource#discard()} method.
-     * In this case, the produced {@link TransactionResult} will be empty.
+     * The user can discard a transaction using the {@link ReactiveTransactionalRedisDataSource#discard()} method.
+     * In this case, the produced {@link TransactionResult} is empty.
+     * If the {@code tx} block completes with a failure, the transaction is discarded automatically and
+     * the resulting {@code Uni} completes with the same failure.
      *
-     * @param tx the consumer receiving the transactional redis data source. The enqueued commands are only executed
-     *        at the end of the block.
+     * @param tx the consumer receiving the transactional Redis data source. The enqueued commands are only executed
+     *        when this block completes with an item.
      * @param watchedKeys the keys to watch during the execution of the transaction. If one of these key is modified before
      *        the completion of the transaction, the transaction is discarded.
      */
     Uni<TransactionResult> withTransaction(Function<ReactiveTransactionalRedisDataSource, Uni<Void>> tx, String... watchedKeys);
 
     /**
-     * Retrieves a {@link RedisDataSource} enqueuing commands in a Redis Transaction ({@code MULTI}).
-     * Note that transaction acquires a single connection, and all the commands are enqueued in this connection.
-     * The commands are only executed when the passed block emits the {@code null} item.
+     * Obtains a {@link ReactiveRedisDataSource} that enqueues commands in a Redis Transaction ({@code WATCH} & {@code MULTI}),
+     * starts watching the given {@code watchedKeys}, passes it to the given {@code preTx} block and if that
+     * doesn't fail, passes it again to the given {@code tx} block. Note that the transaction acquires a single
+     * connection and all the commands are enqueued on this connection. The commands are only executed when
+     * the {@code Uni} returned by {@code tx} produces an item.
      * <p>
-     * This variant also allows executing code before the transaction gets started but after the key being watched:
+     * This variant also allows executing code before the transaction gets started but after the keys are watched:
      *
      * <pre>
-     *     WATCH key
-     *     // preTxBlock
+     * WATCH key
+     *     // preTx
      *     element = ZRANGE k 0 0
-     *     // TxBlock
-     *     MULTI
-     *        ZREM k element
-     *     EXEC
+     * MULTI
+     *     // tx
+     *     ZREM k element
+     * EXEC
      * </pre>
      * <p>
      * The {@code preTxBlock} returns a {@link Uni Uni&lt;I&gt;}. The produced value is received by the {@code tx} block,
      * which can use that value to execute the appropriate operation in the transaction. The produced value can also be
-     * retrieved from the produced {@link OptimisticLockingTransactionResult}. Commands issued in the {@code preTxBlock }
-     * must used the passed (single-connection) {@link ReactiveRedisDataSource} instance.
+     * obtained from the returned {@link OptimisticLockingTransactionResult}. Commands issued in the {@code preTx}
+     * block must use the passed (single-connection) {@link ReactiveRedisDataSource} instance.
      * <p>
-     * If the {@code preTxBlock} throws an exception or emits a failure, the transaction is not executed, and the returned
-     * {@link OptimisticLockingTransactionResult} is empty.
+     * If the {@code preTx} block completes with a failure, all watched keys are {@code UNWATCH}ed and the returned
+     * {@code Uni} completes with the same failure.
      * <p>
      * This construct allows implementing operation relying on optimistic locking.
-     * The results of the commands are retrieved using the produced {@link OptimisticLockingTransactionResult}.
+     * The results of the commands can be obtained from the returned {@link OptimisticLockingTransactionResult}.
+     * Errors are represented as {@code Throwable}s in the {@code OptimisticLockingTransactionResult}.
      * <p>
-     * The user can discard a transaction using the {@link TransactionalRedisDataSource#discard()} method.
-     * In this case, the produced {@link OptimisticLockingTransactionResult} will be empty.
-     *
-     * @param tx the consumer receiving the transactional redis data source. The enqueued commands are only executed
-     *        at the end of the block.
+     * The user can discard a transaction using the {@link ReactiveTransactionalRedisDataSource#discard()} method.
+     * In this case, the produced {@link OptimisticLockingTransactionResult} is empty.
+     * If the {@code tx} block completes with a failure, the transaction is discarded automatically and
+     * the resulting {@code Uni} completes with the same failure.
+     *
+     * @param preTx the consumer receiving the Redis data source before the transaction is started but after
+     *        the {@code watchedKeys} are watched.
+     * @param tx the consumer receiving the transactional Redis data source after the transaction is started.
+     *        The enqueued commands are only executed when this block completes with an item.
      * @param watchedKeys the keys to watch during the execution of the transaction. If one of these key is modified before
      *        the completion of the transaction, the transaction is discarded.
      */
-    <I> Uni<OptimisticLockingTransactionResult<I>> withTransaction(Function<ReactiveRedisDataSource, Uni<I>> preTxBlock,
+    <I> Uni<OptimisticLockingTransactionResult<I>> withTransaction(Function<ReactiveRedisDataSource, Uni<I>> preTx,
             BiFunction<I, ReactiveTransactionalRedisDataSource, Uni<Void>> tx,
             String... watchedKeys);