Refactor to Client-Manager-Connection layers

babushka-core/src/socket_listener.rs

@@ -210,7 +210,7 @@ async fn write_result(
                     error_message.into(),
                 ))
             } else {
-                log_warn("received error", error_message.as_str());
+                log_warn("received error", format!("{} for callback {}", error_message.as_str(), callback_index));
                 let mut request_error = response::RequestError::default();
                 if err.is_connection_dropped() {
                     request_error.type_ = response::RequestErrorType::Disconnect.into();
@@ -476,7 +476,7 @@ fn handle_request(request: RedisRequest, client: Client, writer: Rc<Writer>) {
                 }
             },
             None => Err(ClienUsageError::InternalError(
-                "Received empty request".to_string(),
+                format!("Received empty request for callback {}", request.callback_idx),
             )),
         };
 

java/.gitignore

@@ -4,5 +4,5 @@
 # Ignore Gradle build output directory
 build
 
-# Ignore generated files (e.g. protobuf)
-generated
+# Ignore protobuf generated files
+protobuf

java/README.md

@@ -6,11 +6,11 @@ language api to enhance performance and limit cpu cycles at scale.
 
 ## Organization
 
-The Java client (javababushka) contains the following parts:
+The Java client contains the following parts:
 
 1. A Java client (lib folder): wrapper to rust-client.
-2. An examples script: to sanity test javababushka and similar java-clients against a redis host.
-3. A benchmark app: to performance benchmark test javababushka and similar java-clients against a redis host.
+2. An examples script: to sanity test babushka and similar java-clients against a redis host.
+3. A benchmark app: to performance benchmark test babushka and similar java-clients against a redis host.
 
 ## Building
 

java/benchmarks/build.gradle

@@ -15,8 +15,6 @@ dependencies {
     // Use JUnit test framework.
     testImplementation 'org.junit.jupiter:junit-jupiter:5.9.2'
 
-    // This dependency is used internally, and not exposed to consumers on their own compile classpath.
-    implementation 'com.google.guava:guava:32.1.1-jre'
     implementation 'redis.clients:jedis:4.4.3'
     implementation 'io.lettuce:lettuce-core:6.2.6.RELEASE'
     implementation 'commons-cli:commons-cli:1.5.0'
@@ -36,7 +34,7 @@ java {
 
 application {
     // Define the main class for the application.
-    mainClass = 'javababushka.benchmarks.BenchmarkingApp'
+    mainClass = 'babushka.benchmarks.BenchmarkingApp'
     applicationDefaultJvmArgs += "-Djava.library.path=${projectDir}/../target/release:${projectDir}/../target/debug"
 }
 

java/benchmarks/src/main/java/javababushka/benchmarks/BenchmarkingApp.java → java/benchmarks/src/main/java/babushka/benchmarks/BenchmarkingApp.java

@@ -1,16 +1,15 @@
-package javababushka.benchmarks;
-
-import static javababushka.benchmarks.utils.Benchmarking.testClientSetGet;
-
+package babushka.benchmarks;
+
+import babushka.benchmarks.clients.babushka.JniNettyClient;
+import babushka.benchmarks.clients.jedis.JedisClient;
+import babushka.benchmarks.clients.jedis.JedisPseudoAsyncClient;
+import babushka.benchmarks.clients.lettuce.LettuceAsyncClient;
+import babushka.benchmarks.clients.lettuce.LettuceAsyncClusterClient;
+import babushka.benchmarks.clients.lettuce.LettuceClient;
+import babushka.benchmarks.utils.Benchmarking;
 import java.util.Arrays;
 import java.util.Optional;
 import java.util.stream.Stream;
-import javababushka.benchmarks.clients.babushka.JniNettyClient;
-import javababushka.benchmarks.clients.jedis.JedisClient;
-import javababushka.benchmarks.clients.jedis.JedisPseudoAsyncClient;
-import javababushka.benchmarks.clients.lettuce.LettuceAsyncClient;
-import javababushka.benchmarks.clients.lettuce.LettuceAsyncClusterClient;
-import javababushka.benchmarks.clients.lettuce.LettuceClient;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.DefaultParser;
@@ -36,7 +35,7 @@ public static void main(String[] args) {
       // generate the help statement
       if (line.hasOption("help")) {
         HelpFormatter formatter = new HelpFormatter();
-        formatter.printHelp("javababushka", options);
+        formatter.printHelp("babushka", options);
         return;
       }
 
@@ -50,27 +49,27 @@ public static void main(String[] args) {
       switch (client) {
         case JEDIS:
           // run testClientSetGet on JEDIS sync client
-          testClientSetGet(JedisClient::new, runConfiguration, false);
+          Benchmarking.testClientSetGet(JedisClient::new, runConfiguration, false);
           break;
         case JEDIS_ASYNC:
           // run testClientSetGet on JEDIS pseudo-async client
-          testClientSetGet(JedisPseudoAsyncClient::new, runConfiguration, true);
+          Benchmarking.testClientSetGet(JedisPseudoAsyncClient::new, runConfiguration, true);
           break;
         case LETTUCE:
-          testClientSetGet(LettuceClient::new, runConfiguration, false);
+          Benchmarking.testClientSetGet(LettuceClient::new, runConfiguration, false);
           break;
         case LETTUCE_ASYNC:
           if (runConfiguration.clusterModeEnabled) {
-            testClientSetGet(LettuceAsyncClusterClient::new, runConfiguration, true);
+            Benchmarking.testClientSetGet(LettuceAsyncClusterClient::new, runConfiguration, true);
           } else {
-            testClientSetGet(LettuceAsyncClient::new, runConfiguration, true);
+            Benchmarking.testClientSetGet(LettuceAsyncClient::new, runConfiguration, true);
           }
           break;
         case BABUSHKA:
-          testClientSetGet(() -> new JniNettyClient(false), runConfiguration, false);
+          Benchmarking.testClientSetGet(() -> new JniNettyClient(false), runConfiguration, false);
           break;
         case BABUSHKA_ASYNC:
-          testClientSetGet(() -> new JniNettyClient(true), runConfiguration, true);
+          Benchmarking.testClientSetGet(() -> new JniNettyClient(true), runConfiguration, true);
           break;
       }
     }

java/benchmarks/src/main/java/javababushka/benchmarks/clients/AsyncClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/AsyncClient.java

@@ -1,8 +1,8 @@
-package javababushka.benchmarks.clients;
+package babushka.benchmarks.clients;
 
+import babushka.benchmarks.utils.ConnectionSettings;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import javababushka.benchmarks.utils.ConnectionSettings;
 
 /** A Redis client with async capabilities */
 public interface AsyncClient<T> extends Client {
@@ -13,7 +13,7 @@ public interface AsyncClient<T> extends Client {
 
   Future<T> asyncSet(String key, String value);
 
-  Future<String> asyncGet(String key);
+  Future<T> asyncGet(String key);
 
   default <T> T waitForResult(Future<T> future) {
     return waitForResult(future, DEFAULT_TIMEOUT_MILLISECOND);

java/benchmarks/src/main/java/javababushka/benchmarks/clients/Client.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/Client.java

@@ -1,6 +1,6 @@
-package javababushka.benchmarks.clients;
+package babushka.benchmarks.clients;
 
-import javababushka.benchmarks.utils.ConnectionSettings;
+import babushka.benchmarks.utils.ConnectionSettings;
 
 /** A Redis client interface */
 public interface Client {

java/benchmarks/src/main/java/javababushka/benchmarks/clients/SyncClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/SyncClient.java

@@ -1,4 +1,4 @@
-package javababushka.benchmarks.clients;
+package babushka.benchmarks.clients;
 
 /** A Redis client with sync capabilities */
 public interface SyncClient extends Client {

java/benchmarks/src/main/java/babushka/benchmarks/clients/babushka/JniNettyClient.java

@@ -0,0 +1,68 @@
+package babushka.benchmarks.clients.babushka;
+
+import babushka.api.Awaiter;
+import babushka.api.Client;
+import babushka.api.Commands;
+import babushka.api.Connection;
+import babushka.benchmarks.clients.AsyncClient;
+import babushka.benchmarks.clients.SyncClient;
+import babushka.benchmarks.utils.ConnectionSettings;
+import java.util.concurrent.Future;
+
+public class JniNettyClient implements SyncClient, AsyncClient {
+
+  private final Connection connection;
+  private final Commands asyncCommands;
+
+  private String name = "JNI Netty";
+
+  public JniNettyClient(boolean async) {
+    name += async ? " async" : " sync";
+    connection = Client.CreateConnection();
+    asyncCommands = Client.GetAsyncCommands(connection);
+  }
+
+  @Override
+  public String getName() {
+    return name;
+  }
+
+  @Override
+  public void connectToRedis() {
+    connectToRedis(new ConnectionSettings("localhost", 6379, false, false));
+  }
+
+  @Override
+  public void connectToRedis(ConnectionSettings connectionSettings) {
+    waitForResult(asyncConnectToRedis(connectionSettings));
+  }
+
+  @Override
+  public Future<Boolean> asyncConnectToRedis(ConnectionSettings connectionSettings) {
+    return connection.connectToRedis(
+        connectionSettings.host,
+        connectionSettings.port,
+        connectionSettings.useSsl,
+        connectionSettings.clusterMode);
+  }
+
+  @Override
+  public Future<String> asyncSet(String key, String value) {
+    return asyncCommands.set(key, value);
+  }
+
+  @Override
+  public Future<String> asyncGet(String key) {
+    return asyncCommands.get(key);
+  }
+
+  @Override
+  public void set(String key, String value) {
+    Awaiter.await(asyncCommands.set(key, value));
+  }
+
+  @Override
+  public String get(String key) {
+    return Awaiter.await(asyncCommands.get(key));
+  }
+}

java/benchmarks/src/main/java/javababushka/benchmarks/clients/jedis/JedisClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/jedis/JedisClient.java

@@ -1,7 +1,7 @@
-package javababushka.benchmarks.clients.jedis;
+package babushka.benchmarks.clients.jedis;
 
-import javababushka.benchmarks.clients.SyncClient;
-import javababushka.benchmarks.utils.ConnectionSettings;
+import babushka.benchmarks.clients.SyncClient;
+import babushka.benchmarks.utils.ConnectionSettings;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
 

java/benchmarks/src/main/java/javababushka/benchmarks/clients/jedis/JedisPseudoAsyncClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/jedis/JedisPseudoAsyncClient.java

@@ -1,9 +1,9 @@
-package javababushka.benchmarks.clients.jedis;
+package babushka.benchmarks.clients.jedis;
 
+import babushka.benchmarks.clients.AsyncClient;
+import babushka.benchmarks.utils.ConnectionSettings;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
-import javababushka.benchmarks.clients.AsyncClient;
-import javababushka.benchmarks.utils.ConnectionSettings;
 
 /**
  * A Jedis client with pseudo-async capabilities. Jedis doesn't provide async API

java/benchmarks/src/main/java/javababushka/benchmarks/clients/lettuce/LettuceAsyncClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/lettuce/LettuceAsyncClient.java

@@ -1,14 +1,14 @@
-package javababushka.benchmarks.clients.lettuce;
+package babushka.benchmarks.clients.lettuce;
 
+import babushka.benchmarks.clients.AsyncClient;
+import babushka.benchmarks.utils.ConnectionSettings;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.codec.StringCodec;
 import java.util.concurrent.Future;
-import javababushka.benchmarks.clients.AsyncClient;
-import javababushka.benchmarks.utils.ConnectionSettings;
 
 /** A Lettuce client with async capabilities see: https://lettuce.io/ */
 public class LettuceAsyncClient implements AsyncClient<String> {

java/benchmarks/src/main/java/javababushka/benchmarks/clients/lettuce/LettuceAsyncClusterClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/lettuce/LettuceAsyncClusterClient.java

@@ -1,14 +1,14 @@
 /*
  * This Java source file was generated by the Gradle 'init' task.
  */
-package javababushka.benchmarks.clients.lettuce;
+package babushka.benchmarks.clients.lettuce;
 
+import babushka.benchmarks.utils.ConnectionSettings;
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
-import javababushka.benchmarks.utils.ConnectionSettings;
 
 public class LettuceAsyncClusterClient extends LettuceAsyncClient {
 

java/benchmarks/src/main/java/javababushka/benchmarks/clients/lettuce/LettuceClient.java → java/benchmarks/src/main/java/babushka/benchmarks/clients/lettuce/LettuceClient.java

@@ -1,10 +1,10 @@
-package javababushka.benchmarks.clients.lettuce;
+package babushka.benchmarks.clients.lettuce;
 
+import babushka.benchmarks.clients.SyncClient;
+import babushka.benchmarks.utils.ConnectionSettings;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.sync.RedisStringCommands;
-import javababushka.benchmarks.clients.SyncClient;
-import javababushka.benchmarks.utils.ConnectionSettings;
 
 /** A Lettuce client with sync capabilities see: https://lettuce.io/ */
 public class LettuceClient implements SyncClient {

java/benchmarks/src/main/java/javababushka/benchmarks/utils/Benchmarking.java → java/benchmarks/src/main/java/babushka/benchmarks/utils/Benchmarking.java

@@ -1,5 +1,9 @@
-package javababushka.benchmarks.utils;
+package babushka.benchmarks.utils;
 
+import babushka.benchmarks.BenchmarkingApp;
+import babushka.benchmarks.clients.AsyncClient;
+import babushka.benchmarks.clients.Client;
+import babushka.benchmarks.clients.SyncClient;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -11,10 +15,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
-import javababushka.benchmarks.BenchmarkingApp;
-import javababushka.benchmarks.clients.AsyncClient;
-import javababushka.benchmarks.clients.Client;
-import javababushka.benchmarks.clients.SyncClient;
 import org.apache.commons.lang3.tuple.Pair;
 
 /** Class to calculate latency on client-actions */
@@ -127,6 +127,7 @@ public static void printResults(
     System.out.printf("Iterations: %d%n", iterations);
     System.out.printf("TPS: %f%n", iterations / duration);
     int totalHits = 0;
+
     for (Map.Entry<ChosenAction, LatencyResults> entry : resultsMap.entrySet()) {
       ChosenAction action = entry.getKey();
       LatencyResults results = entry.getValue();
@@ -140,7 +141,7 @@ public static void printResults(
       System.out.printf("Total hits: %d%n", results.totalHits);
       totalHits += results.totalHits;
     }
-    System.out.println("Total hits: " + totalHits);
+    System.out.printf("%nTotal hits: %d%n", totalHits);
   }
 
   public static void testClientSetGet(

java/benchmarks/src/main/java/javababushka/benchmarks/utils/ChosenAction.java → java/benchmarks/src/main/java/babushka/benchmarks/utils/ChosenAction.java

@@ -1,4 +1,4 @@
-package javababushka.benchmarks.utils;
+package babushka.benchmarks.utils;
 
 public enum ChosenAction {
   GET_NON_EXISTING,

java/benchmarks/src/main/java/javababushka/benchmarks/utils/ConnectionSettings.java → java/benchmarks/src/main/java/babushka/benchmarks/utils/ConnectionSettings.java

@@ -1,4 +1,4 @@
-package javababushka.benchmarks.utils;
+package babushka.benchmarks.utils;
 
 import lombok.AllArgsConstructor;
 

java/benchmarks/src/main/java/javababushka/benchmarks/utils/JsonWriter.java → java/benchmarks/src/main/java/babushka/benchmarks/utils/JsonWriter.java

@@ -1,4 +1,4 @@
-package javababushka.benchmarks.utils;
+package babushka.benchmarks.utils;
 
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;

java/benchmarks/src/main/java/javababushka/benchmarks/utils/LatencyResults.java → java/benchmarks/src/main/java/babushka/benchmarks/utils/LatencyResults.java

@@ -1,4 +1,4 @@
-package javababushka.benchmarks.utils;
+package babushka.benchmarks.utils;
 
 import lombok.AllArgsConstructor;