Add managers layer to client

java/client/build.gradle

@@ -52,16 +52,19 @@ tasks.register('cleanProtobuf') {
 tasks.register('buildRustRelease', Exec) {
     commandLine 'cargo', 'build', '--release'
     workingDir project.rootDir
+    environment 'CARGO_TERM_COLOR', 'always'
 }
 
 tasks.register('buildRustReleaseStrip', Exec) {
     commandLine 'cargo', 'build', '--release', '--strip'
     workingDir project.rootDir
+    environment 'CARGO_TERM_COLOR', 'always'
 }
 
 tasks.register('buildRust', Exec) {
     commandLine 'cargo', 'build'
     workingDir project.rootDir
+    environment 'CARGO_TERM_COLOR', 'always'
 }
 
 tasks.register('buildWithRust') {

java/client/src/main/java/babushka/connectors/ClientState.java

@@ -0,0 +1,24 @@
+package babushka.connectors;
+
+public class ClientState {
+
+  private ClientState() {}
+
+  /**
+   * A read only client state. It is supposed that main Client class will have instance of the state
+   * of this type and won't be able to change the state directly.
+   */
+  public static interface ReadOnlyClientState {
+    /** Check that connection established. This doesn't validate whether it is alive. */
+    boolean isConnected();
+
+    /** Check that connection is not yet established. */
+    boolean isInitializing();
+  }
+
+  /** A client state which accepts switching to <em>Connected</em> state. */
+  public static interface OpenableClientState extends ReadOnlyClientState {
+    /** Report connection status. */
+    void connect(boolean successful);
+  }
+}

java/client/src/main/java/babushka/connectors/handlers/CallbackDispatcher.java

@@ -1,15 +1,30 @@
 package babushka.connectors.handlers;
 
+import babushka.connectors.ClientState;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.tuple.Pair;
 import response.ResponseOuterClass.Response;
 
 /** Holder for resources required to dispatch responses and used by {@link ReadHandler}. */
+@RequiredArgsConstructor
 public class CallbackDispatcher {
-  /** Unique request ID (callback ID). Thread-safe. */
+
+  /** Reserved callback ID for connection request. */
+  private final int CONNECTION_PROMISE_ID = 0;
+
+  /** Client state reference. */
+  private final ClientState.ReadOnlyClientState clientState;
+
+  /**
+   * Unique request ID (callback ID). Thread-safe and overflow-safe.<br>
+   * Note: Protobuf packet contains callback ID as uint32, but it stores data as a bit field.<br>
+   * Negative java values would be shown as positive on rust side. Meanwhile, no data loss happen,
+   * because callback ID remains unique.
+   */
   private final AtomicInteger requestId = new AtomicInteger(0);
 
   /**
@@ -18,28 +33,28 @@ public class CallbackDispatcher {
    */
   private final Map<Integer, CompletableFuture<Response>> responses = new ConcurrentHashMap<>();
 
-  /**
-   * Storage for connection request similar to {@link #responses}. Unfortunately, connection
-   * requests can't be stored in the same storage, because callback ID = 0 is hardcoded for
-   * connection requests.
-   */
-  private final CompletableFuture<Response> connectionPromise = new CompletableFuture<>();
-
   /**
    * Register a new request to be sent. Once response received, the given future completes with it.
    *
    * @return A pair of unique callback ID which should set into request and a client promise for
    *     response.
    */
   public Pair<Integer, CompletableFuture<Response>> registerRequest() {
-    int callbackId = requestId.incrementAndGet();
+    int callbackId = 0;
+    do {
+      callbackId = requestId.getAndIncrement();
+    } while (responses.containsKey(callbackId));
     var future = new CompletableFuture<Response>();
     responses.put(callbackId, future);
     return Pair.of(callbackId, future);
   }
 
   public CompletableFuture<Response> registerConnection() {
-    return connectionPromise;
+    var res = registerRequest();
+    if (res.getKey() != CONNECTION_PROMISE_ID) {
+      throw new IllegalStateException();
+    }
+    return res.getValue();
   }
 
   /**
@@ -48,17 +63,20 @@ public CompletableFuture<Response> registerConnection() {
    * @param response A response received
    */
   public void completeRequest(Response response) {
-    int callbackId = response.getCallbackIdx();
-    if (callbackId == 0) {
-      connectionPromise.completeAsync(() -> response);
+    // A connection response doesn't contain a callback id
+    int callbackId =
+        clientState.isInitializing() ? CONNECTION_PROMISE_ID : response.getCallbackIdx();
+    CompletableFuture<Response> future = responses.get(callbackId);
+    if (future != null) {
+      future.completeAsync(() -> response);
     } else {
-      responses.get(callbackId).completeAsync(() -> response);
-      responses.remove(callbackId);
+      // TODO: log an error.
+      // probably a response was received after shutdown or `registerRequest` call was missing
     }
+    responses.remove(callbackId);
   }
 
   public void shutdownGracefully() {
-    connectionPromise.cancel(false);
     responses.values().forEach(future -> future.cancel(false));
     responses.clear();
   }

java/client/src/main/java/babushka/connectors/handlers/ChannelHandler.java

@@ -1,13 +1,13 @@
 package babushka.connectors.handlers;
 
 import babushka.connectors.resources.Platform;
+import babushka.connectors.resources.ThreadPoolAllocator;
 import connection_request.ConnectionRequestOuterClass.ConnectionRequest;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.unix.DomainSocketAddress;
-import java.util.OptionalInt;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
 import redis_request.RedisRequestOuterClass.RedisRequest;
 import response.ResponseOuterClass.Response;
 
@@ -24,7 +24,7 @@ public ChannelHandler(CallbackDispatcher callbackDispatcher, String socketPath)
     channel =
         new Bootstrap()
             // TODO let user specify the thread pool or pool size as an option
-            .group(Platform.createNettyThreadPool("babushka-channel", OptionalInt.empty()))
+            .group(ThreadPoolAllocator.createNettyThreadPool("babushka-channel", Optional.empty()))
             .channel(Platform.getClientUdsNettyChannelType())
             .handler(new ProtobufSocketChannelInitializer(callbackDispatcher))
             .connect(new DomainSocketAddress(socketPath))
@@ -52,13 +52,9 @@ public CompletableFuture<Response> connect(ConnectionRequest request) {
     return callbackDispatcher.registerConnection();
   }
 
-  private final AtomicBoolean closed = new AtomicBoolean(false);
-
   /** Closes the UDS connection and frees corresponding resources. */
   public void close() {
-    if (closed.compareAndSet(false, true)) {
-      channel.close();
-      callbackDispatcher.shutdownGracefully();
-    }
+    channel.close();
+    callbackDispatcher.shutdownGracefully();
   }
 }

java/client/src/main/java/babushka/connectors/resources/Platform.java

@@ -1,18 +1,10 @@
 package babushka.connectors.resources;
 
-import io.netty.channel.EventLoopGroup;
 import io.netty.channel.epoll.Epoll;
 import io.netty.channel.epoll.EpollDomainSocketChannel;
-import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.kqueue.KQueue;
 import io.netty.channel.kqueue.KQueueDomainSocketChannel;
-import io.netty.channel.kqueue.KQueueEventLoopGroup;
 import io.netty.channel.unix.DomainSocketChannel;
-import io.netty.util.concurrent.DefaultThreadFactory;
-import java.util.Map;
-import java.util.OptionalInt;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Supplier;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
@@ -44,12 +36,6 @@ public static class Capabilities {
   private static final Capabilities capabilities =
       new Capabilities(isKQueueAvailable(), isEPollAvailable(), false, false);
 
-  /**
-   * Thread pools supplied to <em>Netty</em> to perform all async IO.<br>
-   * Map key is supposed to be pool name + thread count as a string concat product.
-   */
-  private static final Map<String, EventLoopGroup> groups = new ConcurrentHashMap<>();
-
   /** Detect <em>kqueue</em> availability. */
   private static boolean isKQueueAvailable() {
     try {
@@ -70,42 +56,6 @@ private static boolean isEPollAvailable() {
     }
   }
 
-  /**
-   * Allocate Netty thread pool required to manage connection. A thread pool could be shared across
-   * multiple connections.
-   *
-   * @return A new thread pool.
-   */
-  public static EventLoopGroup createNettyThreadPool(String prefix, OptionalInt threadLimit) {
-    int threadCount = threadLimit.orElse(Runtime.getRuntime().availableProcessors());
-    if (capabilities.isKQueueAvailable()) {
-      var name = prefix + "-kqueue-elg";
-      return getOrCreate(
-          name + threadCount,
-          () -> new KQueueEventLoopGroup(threadCount, new DefaultThreadFactory(name, true)));
-    } else if (capabilities.isEPollAvailable()) {
-      var name = prefix + "-epoll-elg";
-      return getOrCreate(
-          name + threadCount,
-          () -> new EpollEventLoopGroup(threadCount, new DefaultThreadFactory(name, true)));
-    }
-    // TODO support IO-Uring and NIO
-
-    throw new RuntimeException("Current platform supports no known thread pool types");
-  }
-
-  /**
-   * Get a cached thread pool from {@link #groups} or create a new one by given lambda and cache.
-   */
-  private static EventLoopGroup getOrCreate(String name, Supplier<EventLoopGroup> supplier) {
-    if (groups.containsKey(name)) {
-      return groups.get(name);
-    }
-    var group = supplier.get();
-    groups.put(name, group);
-    return group;
-  }
-
   /**
    * Get a channel class required by Netty to open a client UDS channel.
    *
@@ -120,20 +70,4 @@ public static Class<? extends DomainSocketChannel> getClientUdsNettyChannelType(
     }
     throw new RuntimeException("Current platform supports no known socket types");
   }
-
-  /**
-   * A JVM shutdown hook to be registered. It is responsible for closing connection and freeing
-   * resources. It is recommended to use a class instead of lambda to ensure that it is called.<br>
-   * See {@link Runtime#addShutdownHook}.
-   */
-  private static class ShutdownHook implements Runnable {
-    @Override
-    public void run() {
-      groups.values().forEach(EventLoopGroup::shutdownGracefully);
-    }
-  }
-
-  static {
-    Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook(), "Babushka-shutdown-hook"));
-  }
 }

java/client/src/main/java/babushka/connectors/resources/ThreadPoolAllocator.java

@@ -0,0 +1,72 @@
+package babushka.connectors.resources;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.kqueue.KQueueEventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+
+/** A class responsible to allocating and deallocating shared thread pools. */
+public class ThreadPoolAllocator {
+
+  /**
+   * Thread pools supplied to <em>Netty</em> to perform all async IO.<br>
+   * Map key is supposed to be pool name + thread count as a string concat product.
+   */
+  private static final Map<String, EventLoopGroup> groups = new ConcurrentHashMap<>();
+
+  /**
+   * Allocate (create new or share existing) Netty thread pool required to manage connection. A
+   * thread pool could be shared across multiple connections.
+   *
+   * @return A new thread pool.
+   */
+  public static EventLoopGroup createNettyThreadPool(String prefix, Optional<Integer> threadLimit) {
+    int threadCount = threadLimit.orElse(Runtime.getRuntime().availableProcessors());
+    if (Platform.getCapabilities().isKQueueAvailable()) {
+      String name = prefix + "-kqueue-elg";
+      return getOrCreate(
+          name + threadCount,
+          () -> new KQueueEventLoopGroup(threadCount, new DefaultThreadFactory(name, true)));
+    } else if (Platform.getCapabilities().isEPollAvailable()) {
+      String name = prefix + "-epoll-elg";
+      return getOrCreate(
+          name + threadCount,
+          () -> new EpollEventLoopGroup(threadCount, new DefaultThreadFactory(name, true)));
+    }
+    // TODO support IO-Uring and NIO
+
+    throw new RuntimeException("Current platform supports no known thread pool types");
+  }
+
+  /**
+   * Get a cached thread pool from {@link #groups} or create a new one by given lambda and cache.
+   */
+  private static EventLoopGroup getOrCreate(String name, Supplier<EventLoopGroup> supplier) {
+    if (groups.containsKey(name)) {
+      return groups.get(name);
+    }
+    EventLoopGroup group = supplier.get();
+    groups.put(name, group);
+    return group;
+  }
+
+  /**
+   * A JVM shutdown hook to be registered. It is responsible for closing connection and freeing
+   * resources. It is recommended to use a class instead of lambda to ensure that it is called.<br>
+   * See {@link Runtime#addShutdownHook}.
+   */
+  private static class ShutdownHook implements Runnable {
+    @Override
+    public void run() {
+      groups.values().forEach(EventLoopGroup::shutdownGracefully);
+    }
+  }
+
+  static {
+    Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook(), "Babushka-shutdown-hook"));
+  }
+}

java/client/src/main/java/babushka/ffi/resolvers/RedisValueResolver.java

@@ -0,0 +1,13 @@
+package babushka.ffi.resolvers;
+
+import response.ResponseOuterClass.Response;
+
+public class RedisValueResolver {
+  /**
+   * Resolve a value received from Redis using given C-style pointer.
+   *
+   * @param pointer A memory pointer from {@link Response}
+   * @return A RESP3 value
+   */
+  public static native Object valueFromPointer(long pointer);
+}

java/client/src/main/java/babushka/ffi/resolvers/BabushkaCoreNativeDefinitions.java → java/client/src/main/java/babushka/ffi/resolvers/SocketListenerResolver.java

@@ -1,9 +1,9 @@
 package babushka.ffi.resolvers;
 
-public class BabushkaCoreNativeDefinitions {
-  public static native String startSocketListenerExternal() throws Exception;
+public class SocketListenerResolver {
 
-  public static native Object valueFromPointer(long pointer);
+  /** Make an FFI call to Babushka to open a UDS socket to connect to. */
+  private static native String startSocketListener() throws Exception;
 
   static {
     System.loadLibrary("javababushka");
@@ -16,7 +16,7 @@ public class BabushkaCoreNativeDefinitions {
    */
   public static String getSocket() {
     try {
-      return startSocketListenerExternal();
+      return startSocketListener();
     } catch (Exception | UnsatisfiedLinkError e) {
       System.err.printf("Failed to create a UDS connection: %s%n%n", e);
       throw new RuntimeException(e);