Java: Add transaction requests for standalone and cluster-mode

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

@@ -2,39 +2,36 @@
 package glide.api;
 
 import static glide.ffi.resolvers.SocketListenerResolver.getSocket;
+import static glide.managers.RequestType.GET_STRING;
+import static glide.managers.RequestType.PING;
+import static glide.managers.RequestType.SET_STRING;
 
+import glide.api.commands.ConnectionCommands;
+import glide.api.commands.StringCommands;
+import glide.api.models.commands.SetOptions;
 import glide.api.models.configuration.BaseClientConfiguration;
+import glide.api.models.exceptions.RedisException;
 import glide.connectors.handlers.CallbackDispatcher;
 import glide.connectors.handlers.ChannelHandler;
 import glide.ffi.resolvers.RedisValueResolver;
 import glide.managers.BaseCommandResponseResolver;
 import glide.managers.CommandManager;
 import glide.managers.ConnectionManager;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.function.BiFunction;
 import lombok.AllArgsConstructor;
+import org.apache.commons.lang3.ArrayUtils;
 import response.ResponseOuterClass.Response;
 
 /** Base Client class for Redis */
 @AllArgsConstructor
-public abstract class BaseClient implements AutoCloseable {
+public abstract class BaseClient implements AutoCloseable, StringCommands, ConnectionCommands {
 
     protected final ConnectionManager connectionManager;
     protected final CommandManager commandManager;
 
-    /**
-     * Extracts the response from the Protobuf response and either throws an exception or returns the
-     * appropriate response as an Object
-     *
-     * @param response Redis protobuf message
-     * @return Response Object
-     */
-    protected Object handleObjectResponse(Response response) {
-        // convert protobuf response into Object and then Object into T
-        return new BaseCommandResponseResolver(RedisValueResolver::valueFromPointer).apply(response);
-    }
-
     /**
      * Async request for an async (non-blocking) Redis client.
      *
@@ -66,8 +63,8 @@ protected static <T> CompletableFuture<T> CreateClient(
      * Closes this resource, relinquishing any underlying resources. This method is invoked
      * automatically on objects managed by the try-with-resources statement.
      *
-     * <p>see: <a
-     * href="https://docs.oracle.com/javase/8/docs/api/java/lang/AutoCloseable.html#close--">AutoCloseable::close()</a>
+     * @see <a
+     *     href="https://docs.oracle.com/javase/8/docs/api/java/lang/AutoCloseable.html#close--">AutoCloseable::close()</a>
      */
     @Override
     public void close() throws ExecutionException {
@@ -91,4 +88,115 @@ protected static ConnectionManager buildConnectionManager(ChannelHandler channel
     protected static CommandManager buildCommandManager(ChannelHandler channelHandler) {
         return new CommandManager(channelHandler);
     }
+
+    /**
+     * Extracts the response from the Protobuf response and either throws an exception or returns the
+     * appropriate response as an <code>Object</code>.
+     *
+     * @param response Redis protobuf message
+     * @return Response <code>Object</code>
+     */
+    protected Object handleObjectResponse(Response response) {
+        // convert protobuf response into Object
+        return new BaseCommandResponseResolver(RedisValueResolver::valueFromPointer).apply(response);
+    }
+
+    /**
+     * Checks that the Response is empty.
+     *
+     * @return An empty response
+     */
+    protected Void handleVoidResponse(Response response) {
+        Object value = handleObjectResponse(response);
+        if (value == null) {
+            return null;
+        }
+        throw new RedisException(
+                "Unexpected return type from Redis: got "
+                        + value.getClass().getSimpleName()
+                        + " expected null");
+    }
+
+    /**
+     * Extracts the response value from the Redis response and either throws an exception or returns
+     * the value as a <code>String</code>.
+     *
+     * @param response Redis protobuf message
+     * @return Response as a <code>String</code>
+     */
+    protected String handleStringResponse(Response response) {
+        Object value = handleObjectResponse(response);
+        if (value instanceof String) {
+            return (String) value;
+        }
+        throw new RedisException(
+                "Unexpected return type from Redis: got "
+                        + value.getClass().getSimpleName()
+                        + " expected String");
+    }
+
+    /**
+     * Extracts the response value from the Redis response and either throws an exception or returns
+     * the value as an <code>Object[]</code>.
+     *
+     * @param response Redis protobuf message
+     * @return Response as an <code>Object[]</code>
+     */
+    protected Object[] handleArrayResponse(Response response) {
+        Object value = handleObjectResponse(response);
+        if (value instanceof Object[]) {
+            return (Object[]) value;
+        }
+        throw new RedisException(
+                "Unexpected return type from Redis: got "
+                        + value.getClass().getSimpleName()
+                        + " expected Object[]");
+    }
+
+    /**
+     * Extracts the response value from the Redis response and either throws an exception or returns
+     * the value as a <code>Map</code>.
+     *
+     * @param response Redis protobuf message
+     * @return Response as a <code>Map</code>.
+     */
+    @SuppressWarnings("unchecked")
+    protected Map<Object, Object> handleMapResponse(Response response) {
+        Object value = handleObjectResponse(response);
+        if (value instanceof Map) {
+            return (Map<Object, Object>) value;
+        }
+        throw new RedisException(
+                "Unexpected return type from Redis: got "
+                        + value.getClass().getSimpleName()
+                        + " expected Map");
+    }
+
+    @Override
+    public CompletableFuture<String> ping() {
+        return commandManager.submitNewCommand(PING, new String[0], this::handleStringResponse);
+    }
+
+    @Override
+    public CompletableFuture<String> ping(String msg) {
+        return commandManager.submitNewCommand(PING, new String[] {msg}, this::handleStringResponse);
+    }
+
+    @Override
+    public CompletableFuture<String> get(String key) {
+        return commandManager.submitNewCommand(
+                GET_STRING, new String[] {key}, this::handleStringResponse);
+    }
+
+    @Override
+    public CompletableFuture<Void> set(String key, String value) {
+        return commandManager.submitNewCommand(
+                SET_STRING, new String[] {key, value}, this::handleVoidResponse);
+    }
+
+    @Override
+    public CompletableFuture<String> set(String key, String value, SetOptions options) {
+        String[] arguments = ArrayUtils.addAll(new String[] {key, value}, options.toArgs());
+        return commandManager.submitNewCommand(SET_STRING, arguments, this::handleStringResponse);
+    }
 }

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

@@ -1,18 +1,27 @@
 /** Copyright GLIDE-for-Redis Project Contributors - SPDX Identifier: Apache-2.0 */
 package glide.api;
 
+import static glide.managers.RequestType.CUSTOM_COMMAND;
+import static glide.managers.RequestType.INFO;
+
 import glide.api.commands.BaseCommands;
+import glide.api.commands.ConnectionCommands;
+import glide.api.commands.ServerCommands;
+import glide.api.models.Transaction;
+import glide.api.models.commands.InfoOptions;
 import glide.api.models.configuration.RedisClientConfiguration;
+import glide.connectors.handlers.ChannelHandler;
 import glide.managers.CommandManager;
 import glide.managers.ConnectionManager;
-import glide.managers.models.Command;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 /**
  * Async (non-blocking) client for Redis in Standalone mode. Use {@link #CreateClient} to request a
  * client to Redis.
  */
-public class RedisClient extends BaseClient implements BaseCommands {
+public class RedisClient extends BaseClient
+        implements BaseCommands, ConnectionCommands, ServerCommands {
 
     protected RedisClient(ConnectionManager connectionManager, CommandManager commandManager) {
         super(connectionManager, commandManager);
@@ -22,16 +31,43 @@ protected RedisClient(ConnectionManager connectionManager, CommandManager comman
      * Async request for an async (non-blocking) Redis client in Standalone mode.
      *
      * @param config Redis client Configuration
-     * @return a Future to connect and return a RedisClient
+     * @return A Future to connect and return a RedisClient
      */
     public static CompletableFuture<RedisClient> CreateClient(RedisClientConfiguration config) {
-        return CreateClient(config, RedisClient::new);
+        try {
+            ChannelHandler channelHandler = buildChannelHandler();
+            ConnectionManager connectionManager = buildConnectionManager(channelHandler);
+            CommandManager commandManager = buildCommandManager(channelHandler);
+            // TODO: Support exception throwing, including interrupted exceptions
+            return connectionManager
+                    .connectToRedis(config)
+                    .thenApply(ignore -> new RedisClient(connectionManager, commandManager));
+        } catch (InterruptedException e) {
+            // Something bad happened while we were establishing netty connection to UDS
+            var future = new CompletableFuture<RedisClient>();
+            future.completeExceptionally(e);
+            return future;
+        }
     }
 
     @Override
     public CompletableFuture<Object> customCommand(String[] args) {
-        Command command =
-                Command.builder().requestType(Command.RequestType.CUSTOM_COMMAND).arguments(args).build();
-        return commandManager.submitNewCommand(command, this::handleObjectResponse);
+        return commandManager.submitNewCommand(CUSTOM_COMMAND, args, this::handleStringResponse);
+    }
+
+    @Override
+    public CompletableFuture<Object[]> exec(Transaction transaction) {
+        return commandManager.submitNewCommand(
+                transaction, Optional.empty(), this::handleArrayResponse);
+    }
+
+    @Override
+    public CompletableFuture<String> info() {
+        return commandManager.submitNewCommand(INFO, new String[0], this::handleStringResponse);
+    }
+
+    @Override
+    public CompletableFuture<String> info(InfoOptions options) {
+        return commandManager.submitNewCommand(INFO, options.toArgs(), this::handleStringResponse);
     }
 }

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

@@ -1,21 +1,29 @@
 /** Copyright GLIDE-for-Redis Project Contributors - SPDX Identifier: Apache-2.0 */
 package glide.api;
 
+import static glide.managers.RequestType.CUSTOM_COMMAND;
+import static glide.managers.RequestType.INFO;
+
 import glide.api.commands.ClusterBaseCommands;
+import glide.api.commands.ClusterServerCommands;
+import glide.api.models.ClusterTransaction;
 import glide.api.models.ClusterValue;
+import glide.api.models.commands.InfoOptions;
 import glide.api.models.configuration.RedisClusterClientConfiguration;
 import glide.api.models.configuration.RequestRoutingConfiguration.Route;
+import glide.connectors.handlers.ChannelHandler;
 import glide.managers.CommandManager;
 import glide.managers.ConnectionManager;
-import glide.managers.models.Command;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 /**
  * Async (non-blocking) client for Redis in Cluster mode. Use {@link #CreateClient} to request a
  * client to Redis.
  */
-public class RedisClusterClient extends BaseClient implements ClusterBaseCommands {
+public class RedisClusterClient extends BaseClient
+        implements ClusterBaseCommands, ClusterServerCommands {
 
     protected RedisClusterClient(ConnectionManager connectionManager, CommandManager commandManager) {
         super(connectionManager, commandManager);
@@ -25,37 +33,94 @@ protected RedisClusterClient(ConnectionManager connectionManager, CommandManager
      * Async request for an async (non-blocking) Redis client in Cluster mode.
      *
      * @param config Redis cluster client Configuration
-     * @return a Future to connect and return a ClusterClient
+     * @return A Future to connect and return a RedisClusterClient
      */
     public static CompletableFuture<RedisClusterClient> CreateClient(
             RedisClusterClientConfiguration config) {
-        return CreateClient(config, RedisClusterClient::new);
+        try {
+            ChannelHandler channelHandler = buildChannelHandler();
+            ConnectionManager connectionManager = buildConnectionManager(channelHandler);
+            CommandManager commandManager = buildCommandManager(channelHandler);
+            // TODO: Support exception throwing, including interrupted exceptions
+            return connectionManager
+                    .connectToRedis(config)
+                    .thenApply(ignored -> new RedisClusterClient(connectionManager, commandManager));
+        } catch (InterruptedException e) {
+            // Something bad happened while we were establishing netty connection to UDS
+            var future = new CompletableFuture<RedisClusterClient>();
+            future.completeExceptionally(e);
+            return future;
+        }
     }
 
     @Override
     public CompletableFuture<ClusterValue<Object>> customCommand(String[] args) {
-        Command command =
-                Command.builder().requestType(Command.RequestType.CUSTOM_COMMAND).arguments(args).build();
         // TODO if a command returns a map as a single value, ClusterValue misleads user
         return commandManager.submitNewCommand(
-                command, response -> ClusterValue.of(handleObjectResponse(response)));
+                CUSTOM_COMMAND,
+                args,
+                Optional.empty(),
+                response -> ClusterValue.of(handleObjectResponse(response)));
     }
 
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<ClusterValue<Object>> customCommand(String[] args, Route route) {
-        Command command =
-                Command.builder()
-                        .requestType(Command.RequestType.CUSTOM_COMMAND)
-                        .arguments(args)
-                        .route(route)
-                        .build();
-
         return commandManager.submitNewCommand(
-                command,
+                CUSTOM_COMMAND,
+                args,
+                Optional.ofNullable(route),
                 response ->
                         route.isSingleNodeRoute()
                                 ? ClusterValue.ofSingleValue(handleObjectResponse(response))
                                 : ClusterValue.ofMultiValue((Map<String, Object>) handleObjectResponse(response)));
     }
+
+    @Override
+    public CompletableFuture<Object[]> exec(ClusterTransaction transaction) {
+        return commandManager.submitNewCommand(
+                transaction, Optional.empty(), this::handleArrayResponse);
+    }
+
+    @Override
+    public CompletableFuture<Object[]> exec(ClusterTransaction transaction, Route route) {
+        return commandManager.submitNewCommand(
+                transaction, Optional.ofNullable(route), this::handleArrayResponse);
+    }
+
+    @Override
+    public CompletableFuture<ClusterValue<String>> info() {
+        return commandManager.submitNewCommand(
+                INFO,
+                new String[0],
+                Optional.empty(),
+                response -> ClusterValue.of(handleStringResponse(response)));
+    }
+
+    @Override
+    public CompletableFuture<ClusterValue<String>> info(Route route) {
+        return commandManager.submitNewCommand(
+                INFO,
+                new String[0],
+                Optional.ofNullable(route),
+                response -> ClusterValue.of(handleStringResponse(response)));
+    }
+
+    @Override
+    public CompletableFuture<ClusterValue<String>> info(InfoOptions options) {
+        return commandManager.submitNewCommand(
+                INFO,
+                options.toArgs(),
+                Optional.empty(),
+                response -> ClusterValue.of(handleStringResponse(response)));
+    }
+
+    @Override
+    public CompletableFuture<ClusterValue<String>> info(InfoOptions options, Route route) {
+        return commandManager.submitNewCommand(
+                INFO,
+                options.toArgs(),
+                Optional.ofNullable(route),
+                response -> ClusterValue.of(handleStringResponse(response)));
+    }
 }