Java: Add WATCH and UNWATCH command

glide-core/src/protobuf/redis_request.proto

@@ -208,6 +208,8 @@ enum RequestType {
     BitFieldReadOnly = 173;
     Move = 174;
     SInterCard = 175;
+    Watch = 176;
+    UnWatch = 177;
 }
 
 message Command {

glide-core/src/request_type.rs

@@ -178,6 +178,8 @@ pub enum RequestType {
     BitFieldReadOnly = 173,
     Move = 174,
     SInterCard = 175,
+    Watch = 176,
+    UnWatch = 177,
 }
 
 fn get_two_word_command(first: &str, second: &str) -> Cmd {
@@ -359,6 +361,8 @@ impl From<::protobuf::EnumOrUnknown<ProtobufRequestType>> for RequestType {
             ProtobufRequestType::Move => RequestType::Move,
             ProtobufRequestType::SInterCard => RequestType::SInterCard,
             ProtobufRequestType::Sort => RequestType::Sort,
+            ProtobufRequestType::Watch => RequestType::Watch,
+            ProtobufRequestType::UnWatch => RequestType::UnWatch,
         }
     }
 }
@@ -536,6 +540,8 @@ impl RequestType {
             RequestType::Move => Some(cmd("MOVE")),
             RequestType::SInterCard => Some(cmd("SINTERCARD")),
             RequestType::Sort => Some(cmd("SORT")),
+            RequestType::Watch => Some(cmd("WATCH")),
+            RequestType::UnWatch => Some(cmd("UNWATCH")),
         }
     }
 }

java/client/src/main/java/glide/api/BaseClient.java

@@ -110,7 +110,9 @@
 import static redis_request.RedisRequestOuterClass.RequestType.TTL;
 import static redis_request.RedisRequestOuterClass.RequestType.Touch;
 import static redis_request.RedisRequestOuterClass.RequestType.Type;
+import static redis_request.RedisRequestOuterClass.RequestType.UnWatch;
 import static redis_request.RedisRequestOuterClass.RequestType.Unlink;
+import static redis_request.RedisRequestOuterClass.RequestType.Watch;
 import static redis_request.RedisRequestOuterClass.RequestType.XAdd;
 import static redis_request.RedisRequestOuterClass.RequestType.XDel;
 import static redis_request.RedisRequestOuterClass.RequestType.XLen;
@@ -153,6 +155,7 @@
 import glide.api.commands.SortedSetBaseCommands;
 import glide.api.commands.StreamBaseCommands;
 import glide.api.commands.StringBaseCommands;
+import glide.api.commands.TransactionsBaseCommands;
 import glide.api.models.Script;
 import glide.api.models.commands.ExpireOptions;
 import glide.api.models.commands.LInsertOptions.InsertPosition;
@@ -214,7 +217,8 @@ public abstract class BaseClient
                 SortedSetBaseCommands,
                 StreamBaseCommands,
                 HyperLogLogBaseCommands,
-                GeospatialIndicesBaseCommands {
+                GeospatialIndicesBaseCommands,
+                TransactionsBaseCommands {
 
     /** Redis simple string response with "OK" */
     public static final String OK = ConstantResponse.OK.toString();
@@ -1733,4 +1737,14 @@ public CompletableFuture<Long> sintercard(@NonNull String[] keys, long limit) {
                         new String[] {SET_LIMIT_REDIS_API, Long.toString(limit)});
         return commandManager.submitNewCommand(SInterCard, arguments, this::handleLongResponse);
     }
+
+    @Override
+    public CompletableFuture<String> watch(@NonNull String[] keys) {
+        return commandManager.submitNewCommand(Watch, keys, this::handleStringResponse);
+    }
+
+    @Override
+    public CompletableFuture<String> unwatch() {
+        return commandManager.submitNewCommand(UnWatch, new String[0], this::handleStringResponse);
+    }
 }

java/client/src/main/java/glide/api/commands/TransactionsBaseCommands.java

@@ -0,0 +1,43 @@
+/** Copyright GLIDE-for-Redis Project Contributors - SPDX Identifier: Apache-2.0 */
+package glide.api.commands;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Supports commands for the "Transactions Commands" group for a standalone and cluster clients.
+ *
+ * @see <a href="https://redis.io/commands/?group=transactions">Transactions Commands</a>
+ */
+public interface TransactionsBaseCommands {
+    /**
+     * Marks the given keys to be watched for conditional execution of a transaction. Transactions
+     * will only execute commands if the watched keys are not modified before execution of the
+     * transaction.
+     *
+     * @apiNote When in cluster mode, all <code>keys</code> must map to the same hash slot.
+     * @see <a href="https://redis.io/docs/latest/commands/watch/">redis.io</a> for details.
+     * @param keys The keys to watch.
+     * @return The string <code>OK</code>.
+     * @example
+     *     <pre>{@code
+     * client.watch("sampleKey");
+     * transaction.set("sampleKey", "foobar");
+     * client.exec(transaction).get(); // Executes successfully and keys are unwatched.
+     * }</pre>
+     */
+    CompletableFuture<String> watch(String[] keys);
+
+    /**
+     * Flushes all the previously watched keys for a transaction. Executing a transaction will
+     * automatically flush all previously watched keys.
+     *
+     * @see <a href="https://redis.io/docs/latest/commands/unwatch/">redis.io</a> for details.
+     * @return The string <code>OK</code>.
+     * @example
+     *     <pre>{@code
+     * client.watch("sampleKey");
+     * client.unwatch(); // Flushes "sampleKey" from watched keys.
+     * }</pre>
+     */
+    CompletableFuture<String> unwatch();
+}

java/client/src/test/java/glide/api/RedisClientTest.java

@@ -160,7 +160,9 @@
 import static redis_request.RedisRequestOuterClass.RequestType.Time;
 import static redis_request.RedisRequestOuterClass.RequestType.Touch;
 import static redis_request.RedisRequestOuterClass.RequestType.Type;
+import static redis_request.RedisRequestOuterClass.RequestType.UnWatch;
 import static redis_request.RedisRequestOuterClass.RequestType.Unlink;
+import static redis_request.RedisRequestOuterClass.RequestType.Watch;
 import static redis_request.RedisRequestOuterClass.RequestType.XAdd;
 import static redis_request.RedisRequestOuterClass.RequestType.XDel;
 import static redis_request.RedisRequestOuterClass.RequestType.XLen;
@@ -5683,4 +5685,47 @@ public void move_returns_success() {
         assertEquals(testResponse, response);
         assertEquals(value, payload);
     }
+
+    @SneakyThrows
+    @Test
+    public void watch_returns_success() {
+        // setup
+        String key1 = "testKey1";
+        String key2 = "testKey2";
+        String[] arguments = new String[] {key1, key2};
+        CompletableFuture<String> testResponse = new CompletableFuture<>();
+        testResponse.complete(OK);
+
+        // match on protobuf request
+        when(commandManager.<String>submitNewCommand(eq(Watch), eq(arguments), any()))
+                .thenReturn(testResponse);
+
+        // exercise
+        CompletableFuture<String> response = service.watch(arguments);
+        String payload = response.get();
+
+        // verify
+        assertEquals(testResponse, response);
+        assertEquals(OK, payload);
+    }
+
+    @SneakyThrows
+    @Test
+    public void unwatch_returns_success() {
+        // setup
+        CompletableFuture<String> testResponse = new CompletableFuture<>();
+        testResponse.complete(OK);
+
+        // match on protobuf request
+        when(commandManager.<String>submitNewCommand(eq(UnWatch), eq(new String[0]), any()))
+                .thenReturn(testResponse);
+
+        // exercise
+        CompletableFuture<String> response = service.unwatch();
+        String payload = response.get();
+
+        // verify
+        assertEquals(testResponse, response);
+        assertEquals(OK, payload);
+    }
 }

java/integTest/src/test/java/glide/SharedCommandTests.java

@@ -31,6 +31,7 @@
 import glide.api.RedisClient;
 import glide.api.RedisClusterClient;
 import glide.api.models.Script;
+import glide.api.models.Transaction;
 import glide.api.models.commands.ConditionalChange;
 import glide.api.models.commands.ExpireOptions;
 import glide.api.models.commands.ListDirection;
@@ -4950,4 +4951,92 @@ public void sintercard(BaseClient client) {
                 assertThrows(ExecutionException.class, () -> client.sintercard(badArr).get());
         assertInstanceOf(RequestException.class, executionException.getCause());
     }
+
+    @SneakyThrows
+    @ParameterizedTest(autoCloseArguments = false)
+    @MethodSource("getClients")
+    public void watch(BaseClient client) {
+        String key1 = "{key}-1" + UUID.randomUUID();
+        String key2 = "{key}-2" + UUID.randomUUID();
+        String key3 = "{key}-3" + UUID.randomUUID();
+        String key4 = "{key}-4" + UUID.randomUUID();
+        String foobarString = "foobar";
+        String helloString = "hello";
+        String[] keys = new String[] {key1, key2, key3};
+        Transaction setFoobarTransaction = new Transaction();
+        Transaction setHelloTransaction = new Transaction();
+        String[] expectedExecResponse = new String[] {OK, OK, OK};
+
+        // Returns null when a watched key is modified before it is executed in a transaction command.
+        // Transaction commands are not performed.
+        assertEquals(OK, standaloneClient.watch(keys).get());
+        assertEquals(OK, standaloneClient.set(key2, helloString).get());
+        setFoobarTransaction.set(key1, foobarString).set(key2, foobarString).set(key3, foobarString);
+        assertEquals(null, standaloneClient.exec(setFoobarTransaction).get());
+        assertEquals(null, standaloneClient.get(key1).get());
+        assertEquals(helloString, standaloneClient.get(key2).get());
+        assertEquals(null, standaloneClient.get(key3).get());
+
+        // Transaction executes command successfully with a read command on the watch key before
+        // transaction is executed.
+        assertEquals(OK, standaloneClient.watch(keys).get());
+        assertEquals(helloString, standaloneClient.get(key2).get());
+        assertArrayEquals(expectedExecResponse, standaloneClient.exec(setFoobarTransaction).get());
+        assertEquals(foobarString, standaloneClient.get(key1).get());
+        assertEquals(foobarString, standaloneClient.get(key2).get());
+        assertEquals(foobarString, standaloneClient.get(key3).get());
+
+        // Transaction executes command successfully with an unmodified watched key
+        assertEquals(OK, standaloneClient.watch(keys).get());
+        assertArrayEquals(expectedExecResponse, standaloneClient.exec(setFoobarTransaction).get());
+        assertEquals(foobarString, standaloneClient.get(key1).get());
+        assertEquals(foobarString, standaloneClient.get(key2).get());
+        assertEquals(foobarString, standaloneClient.get(key3).get());
+
+        // Transaction executes command successfully with a modified watched key but is not in the
+        // transaction.
+        assertEquals(OK, standaloneClient.watch(new String[] {key4}).get());
+        setHelloTransaction.set(key1, helloString).set(key2, helloString).set(key3, helloString);
+        assertArrayEquals(expectedExecResponse, standaloneClient.exec(setHelloTransaction).get());
+        assertEquals(helloString, standaloneClient.get(key1).get());
+        assertEquals(helloString, standaloneClient.get(key2).get());
+        assertEquals(helloString, standaloneClient.get(key3).get());
+
+        // Transaction executes command successfully with an unmodified watched key
+        assertEquals(OK, standaloneClient.watch(keys).get());
+        assertArrayEquals(expectedExecResponse, standaloneClient.exec(setFoobarTransaction).get());
+        assertEquals(foobarString, standaloneClient.get(key1).get());
+        assertEquals(foobarString, standaloneClient.get(key2).get());
+        assertEquals(foobarString, standaloneClient.get(key3).get());
+
+        // WATCH can not have an empty String array parameter
+        ExecutionException executionException =
+                assertThrows(ExecutionException.class, () -> standaloneClient.watch(new String[] {}).get());
+        assertInstanceOf(RequestException.class, executionException.getCause());
+    }
+
+    @SneakyThrows
+    @ParameterizedTest(autoCloseArguments = false)
+    @MethodSource("getClients")
+    public void unwatch(BaseClient client) {
+        String key1 = "{key}-1" + UUID.randomUUID();
+        String key2 = "{key}-2" + UUID.randomUUID();
+        String foobarString = "foobar";
+        String helloString = "hello";
+        String[] keys = new String[] {key1, key2};
+        Transaction setFoobarTransaction = new Transaction();
+        String[] expectedExecResponse = new String[] {OK, OK};
+
+        // UNWATCH returns OK when there no watched keys
+        assertEquals(OK, standaloneClient.unwatch().get());
+
+        // Transaction executes successfully after modifying a watched key then calling UNWATCH
+        assertEquals(OK, standaloneClient.watch(keys).get());
+        assertEquals(OK, standaloneClient.set(key2, helloString).get());
+        assertEquals(OK, standaloneClient.unwatch().get());
+        setFoobarTransaction.set(key1, foobarString).set(key2, foobarString);
+        assertArrayEquals(expectedExecResponse, standaloneClient.exec(setFoobarTransaction).get());
+        assertEquals(foobarString, standaloneClient.get(key1).get());
+        assertEquals(foobarString, standaloneClient.get(key2).get());
+    }
 }

java/integTest/src/test/java/glide/cluster/CommandTests.java

@@ -721,7 +721,8 @@ public static Stream<Arguments> callCrossSlotCommandsWhichShouldFail() {
                         clusterClient.blmove("abc", "def", ListDirection.LEFT, ListDirection.LEFT, 1)),
                 Arguments.of("sintercard", "7.0.0", clusterClient.sintercard(new String[] {"abc", "def"})),
                 Arguments.of(
-                        "sintercard", "7.0.0", clusterClient.sintercard(new String[] {"abc", "def"}, 1)));
+                        "sintercard", "7.0.0", clusterClient.sintercard(new String[] {"abc", "def"}, 1)),
+                Arguments.of("watch", "2.2.0", clusterClient.watch(new String[] {"abc", "def"})));
     }
 
     @SneakyThrows