X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/be0993c5d8171aff9190e152b095677be4088112..6f4d538832b0ee1a4bb45de5d96c9afd5d07f463:/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java

diff --git a/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java b/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
index f3784a2c..16f5fbd0 100644
--- a/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
+++ b/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
@@ -13,7 +13,10 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.Objects;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.StreamSupport;
@@ -24,41 +27,107 @@ public class JsonRpcReader {
 
     private final JsonRpcSender jsonRpcSender;
     private final ObjectMapper objectMapper;
+    private final InputStream input;
     private final Supplier<String> lineSupplier;
 
-    public JsonRpcReader(
-            final JsonRpcSender jsonRpcSender, final Supplier<String> lineSupplier
-    ) {
+    public JsonRpcReader(final JsonRpcSender jsonRpcSender, final Supplier<String> lineSupplier) {
         this.jsonRpcSender = jsonRpcSender;
+        this.input = null;
         this.lineSupplier = lineSupplier;
         this.objectMapper = Util.createJsonObjectMapper();
     }
 
+    public JsonRpcReader(final JsonRpcSender jsonRpcSender, final InputStream input) {
+        this.jsonRpcSender = jsonRpcSender;
+        this.input = input;
+        this.lineSupplier = null;
+        this.objectMapper = Util.createJsonObjectMapper();
+    }
+
     public void readMessages(final RequestHandler requestHandler, final Consumer<JsonRpcResponse> responseHandler) {
-        while (!Thread.interrupted()) {
-            JsonRpcMessage message = readMessage();
-            if (message == null) break;
-
-            if (message instanceof final JsonRpcRequest jsonRpcRequest) {
-                logger.debug("Received json rpc request, method: " + jsonRpcRequest.getMethod());
-                final var response = handleRequest(requestHandler, jsonRpcRequest);
-                if (response != null) {
-                    jsonRpcSender.sendResponse(response);
+        if (input != null) {
+            JsonRpcMessage message = parseJsonRpcMessage(input);
+            if (message == null) {
+                return;
+            }
+
+            handleMessage(message, requestHandler, responseHandler);
+            return;
+        }
+
+        final var executor = Executors.newFixedThreadPool(10);
+        try {
+            while (!Thread.interrupted()) {
+                final var input = lineSupplier.get();
+                if (input == null) {
+                    logger.trace("Reached end of JSON-RPC input stream.");
+                    break;
                 }
-            } else if (message instanceof JsonRpcResponse jsonRpcResponse) {
-                responseHandler.accept(jsonRpcResponse);
-            } else {
-                final var responseList = ((JsonRpcBatchMessage) message).getMessages().stream().map(jsonNode -> {
+
+                logger.trace("Incoming JSON-RPC message: {}", input);
+                final var message = parseJsonRpcMessage(input);
+                if (message == null) {
+                    continue;
+                }
+
+                executor.submit(() -> handleMessage(message, requestHandler, responseHandler));
+            }
+        } finally {
+            Util.closeExecutorService(executor);
+        }
+    }
+
+    private void handleMessage(
+            final JsonRpcMessage message,
+            final RequestHandler requestHandler,
+            final Consumer<JsonRpcResponse> responseHandler
+    ) {
+        if (message instanceof final JsonRpcRequest jsonRpcRequest) {
+            logger.debug("Received json rpc request, method: " + jsonRpcRequest.getMethod());
+            final var response = handleRequest(requestHandler, jsonRpcRequest);
+            if (response != null) {
+                jsonRpcSender.sendResponse(response);
+            }
+        } else if (message instanceof JsonRpcResponse jsonRpcResponse) {
+            responseHandler.accept(jsonRpcResponse);
+        } else {
+            final var messages = ((JsonRpcBatchMessage) message).getMessages();
+            final var responseList = new ArrayList<JsonRpcResponse>(messages.size());
+            final var executor = Executors.newFixedThreadPool(10);
+            try {
+                final var lock = new ReentrantLock();
+                messages.forEach(jsonNode -> {
                     final JsonRpcRequest request;
                     try {
                         request = parseJsonRpcRequest(jsonNode);
                     } catch (JsonRpcException e) {
-                        return JsonRpcResponse.forError(e.getError(), getId(jsonNode));
+                        final var response = JsonRpcResponse.forError(e.getError(), getId(jsonNode));
+                        lock.lock();
+                        try {
+                            responseList.add(response);
+                        } finally {
+                            lock.unlock();
+                        }
+                        return;
                     }
 
-                    return handleRequest(requestHandler, request);
-                }).filter(Objects::nonNull).toList();
+                    executor.submit(() -> {
+                        final var response = handleRequest(requestHandler, request);
+                        if (response != null) {
+                            lock.lock();
+                            try {
+                                responseList.add(response);
+                            } finally {
+                                lock.unlock();
+                            }
+                        }
+                    });
+                });
+            } finally {
+                Util.closeExecutorService(executor);
+            }
 
+            if (responseList.size() > 0) {
                 jsonRpcSender.sendBatchResponses(responseList);
             }
         }
@@ -85,26 +154,23 @@ public class JsonRpcReader {
         return null;
     }
 
-    private JsonRpcMessage readMessage() {
-        while (!Thread.interrupted()) {
-            String input = lineSupplier.get();
-
-            if (input == null) {
-                logger.trace("Reached end of JSON-RPC input stream.");
-                break;
-            }
-
-            logger.trace("Incoming JSON-RPC message: {}", input);
-            JsonRpcMessage message = parseJsonRpcMessage(input);
-            if (message == null) continue;
-
-            return message;
+    private JsonRpcMessage parseJsonRpcMessage(final String input) {
+        final JsonNode jsonNode;
+        try {
+            jsonNode = objectMapper.readTree(input);
+        } catch (JsonParseException e) {
+            jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.PARSE_ERROR,
+                    e.getMessage(),
+                    null), null));
+            return null;
+        } catch (IOException e) {
+            throw new AssertionError(e);
         }
 
-        return null;
+        return parseJsonRpcMessage(jsonNode);
     }
 
-    private JsonRpcMessage parseJsonRpcMessage(final String input) {
+    private JsonRpcMessage parseJsonRpcMessage(final InputStream input) {
         final JsonNode jsonNode;
         try {
             jsonNode = objectMapper.readTree(input);
@@ -117,6 +183,10 @@ public class JsonRpcReader {
             throw new AssertionError(e);
         }
 
+        return parseJsonRpcMessage(jsonNode);
+    }
+
+    private JsonRpcMessage parseJsonRpcMessage(final JsonNode jsonNode) {
         if (jsonNode == null) {
             jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
                     "invalid request",