Response pump is now faster due to Response parsing being done in another thread instead on the read-loop.

Adrian Todt · Adrian Todt · commit ef91abb5b146 · 2020-04-19T23:27:01.000-03:00
diff --git a/src/main/java/com/rethinkdb/net/DefaultConnectionFactory.java b/src/main/java/com/rethinkdb/net/DefaultConnectionFactory.java
@@ -199,11 +199,17 @@ public ThreadResponsePump(ConnectionSocket socket, boolean daemon) {
 
                     // read response and send it to whoever is waiting, if anyone
                     try {
-                        final Response response = Response.readFromSocket(socket);
-                        final CompletableFuture<Response> awaiter = awaiting.remove(response.token);
-                        if (awaiter != null) {
-                            awaiter.complete(response);
-                        }
+                        CompletableFuture.supplyAsync(Response.readFromSocket(socket)).handle((response, t) -> {
+                            if (t != null) {
+                                shutdown(t);
+                            } else {
+                                final CompletableFuture<Response> awaiter = awaiting.remove(response.token);
+                                if (awaiter != null) {
+                                    awaiter.complete(response);
+                                }
+                            }
+                            return null;
+                        });
                     } catch (Exception e) {
                         shutdown(e);
                         return;
@@ -229,12 +235,12 @@ public boolean isAlive() {
             return thread.isAlive();
         }
 
-        private void shutdown(Exception e) {
+        private void shutdown(Throwable t) {
             Map<Long, CompletableFuture<Response>> awaiting = this.awaiting;
             this.awaiting = null;
             thread.interrupt();
             if (awaiting != null) {
-                awaiting.forEach((token, future) -> future.completeExceptionally(e));
+                awaiting.forEach((token, future) -> future.completeExceptionally(t));
             }
         }
 
diff --git a/src/main/java/com/rethinkdb/net/Response.java b/src/main/java/com/rethinkdb/net/Response.java
@@ -20,6 +20,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 public class Response {
@@ -83,7 +84,7 @@ public String toString() {
     }
 
     @SuppressWarnings("unchecked")
-    public static Response readFromSocket(ConnectionSocket socket) {
+    public static Supplier<Response> readFromSocket(ConnectionSocket socket) {
         final ByteBuffer header = socket.read(12);
         final long token = header.getLong();
         final int responseLength = header.getInt();
@@ -100,20 +101,22 @@ public static Response readFromSocket(ConnectionSocket socket) {
             );
         }
 
-        Map<String, Object> json = Internals.readJson(buffer);
-        return new Response(
-            token,
-            ResponseType.fromValue(((Long) json.get("t")).intValue()),
-            (List<Object>) json.getOrDefault("r", Collections.emptyList()),
-            ((List<Long>) json.getOrDefault("n", Collections.emptyList()))
-                .stream()
-                .map(Long::intValue)
-                .map(ResponseNote::maybeFromValue)
-                .filter(Objects::nonNull)
-                .collect(Collectors.toList()),
-            Profile.fromList((List<Object>) json.get("p")),
-            Backtrace.fromList((List<Object>) json.getOrDefault("b", null)),
-            json.containsKey("e") ? ErrorType.maybeFromValue(((Long) json.get("e")).intValue()) : null
-        );
+        return () -> {
+            Map<String, Object> json = Internals.readJson(buffer);
+            return new Response(
+                token,
+                ResponseType.fromValue(((Long) json.get("t")).intValue()),
+                (List<Object>) json.getOrDefault("r", Collections.emptyList()),
+                ((List<Long>) json.getOrDefault("n", Collections.emptyList()))
+                    .stream()
+                    .map(Long::intValue)
+                    .map(ResponseNote::maybeFromValue)
+                    .filter(Objects::nonNull)
+                    .collect(Collectors.toList()),
+                Profile.fromList((List<Object>) json.get("p")),
+                Backtrace.fromList((List<Object>) json.getOrDefault("b", null)),
+                json.containsKey("e") ? ErrorType.maybeFromValue(((Long) json.get("e")).intValue()) : null
+            );
+        };
     }
 }
