Use pattern matching switch cases

[signal-cli] / 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 27da9b0be5606fb1917e601da6c951350556cf74..f9ef71a1f05cbc8ca72cba4ee90ab88786d5eeba 100644 (file)
--- a/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
+++ b/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
@@ -14,7 +14,9 @@ import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.InputStream;
 
 import java.io.IOException;
 import java.io.InputStream;

-import java.util.Objects;
+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;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.StreamSupport;


@@ -53,20 +55,25 @@ public class JsonRpcReader {
             return;
         }
 
             return;
         }
 

-        while (!Thread.interrupted()) {
-            String input = lineSupplier.get();
-            if (input == null) {
-                logger.trace("Reached end of JSON-RPC input stream.");
-                break;
-            }
+        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;
+                }

 
 

-            logger.trace("Incoming JSON-RPC message: {}", input);
-            JsonRpcMessage message = parseJsonRpcMessage(input);
-            if (message == null) {
-                continue;
-            }
+                logger.trace("Incoming JSON-RPC message: {}", input);
+                final var message = parseJsonRpcMessage(input);
+                if (message == null) {
+                    continue;
+                }

 
 

-            handleMessage(message, requestHandler, responseHandler);
+                executor.submit(() -> handleMessage(message, requestHandler, responseHandler));
+            }
+        } finally {
+            Util.closeExecutorService(executor);

         }
     }
 
         }
     }
 


@@ -75,28 +82,55 @@ public class JsonRpcReader {
             final RequestHandler requestHandler,
             final Consumer<JsonRpcResponse> responseHandler
     ) {
             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);
+        switch (message) {
+            case 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 responseList = ((JsonRpcBatchMessage) message).getMessages().stream().map(jsonNode -> {
-                final JsonRpcRequest request;
+            case JsonRpcResponse jsonRpcResponse -> responseHandler.accept(jsonRpcResponse);
+            case JsonRpcBatchMessage jsonRpcBatchMessage -> {
+                final var messages = jsonRpcBatchMessage.getMessages();
+                final var responseList = new ArrayList<JsonRpcResponse>(messages.size());
+                final var executor = Executors.newFixedThreadPool(10);

                 try {
                 try {

-                    request = parseJsonRpcRequest(jsonNode);
-                } catch (JsonRpcException e) {
-                    return JsonRpcResponse.forError(e.getError(), getId(jsonNode));
+                    final var lock = new ReentrantLock();
+                    messages.forEach(jsonNode -> {
+                        final JsonRpcRequest request;
+                        try {
+                            request = parseJsonRpcRequest(jsonNode);
+                        } catch (JsonRpcException e) {
+                            final var response = JsonRpcResponse.forError(e.getError(), getId(jsonNode));
+                            lock.lock();
+                            try {
+                                responseList.add(response);
+                            } finally {
+                                lock.unlock();
+                            }
+                            return;
+                        }
+
+                        executor.submit(() -> {
+                            final var response = handleRequest(requestHandler, request);
+                            if (response != null) {
+                                lock.lock();
+                                try {
+                                    responseList.add(response);
+                                } finally {
+                                    lock.unlock();
+                                }
+                            }
+                        });
+                    });
+                } finally {
+                    Util.closeExecutorService(executor);

                 }
 
                 }
 

-                return handleRequest(requestHandler, request);
-            }).filter(Objects::nonNull).toList();
-
-            if (responseList.size() > 0) {
-                jsonRpcSender.sendBatchResponses(responseList);
+                if (!responseList.isEmpty()) {
+                    jsonRpcSender.sendBatchResponses(responseList);
+                }

             }
         }
     }
             }
         }
     }


@@ -161,7 +195,7 @@ public class JsonRpcReader {
                     null), null));
             return null;
         } else if (jsonNode.isArray()) {
                     null), null));
             return null;
         } else if (jsonNode.isArray()) {

-            if (jsonNode.size() == 0) {
+            if (jsonNode.isEmpty()) {

                 jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
                         "invalid request",
                         null), null));
                 jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
                         "invalid request",
                         null), null));


@@ -189,7 +223,7 @@ public class JsonRpcReader {
 
     private ValueNode getId(JsonNode jsonNode) {
         final var id = jsonNode.get("id");
 
     private ValueNode getId(JsonNode jsonNode) {
         final var id = jsonNode.get("id");

-        return id instanceof ValueNode ? (ValueNode) id : null;
+        return id instanceof ValueNode value ? value : null;

     }
 
     private JsonRpcRequest parseJsonRpcRequest(final JsonNode input) throws JsonRpcException {
     }
 
     private JsonRpcRequest parseJsonRpcRequest(final JsonNode input) throws JsonRpcException {