Create UNIX sockets JsonRpc client/server

consensusj-jsonrpc-unix/build.gradle

@@ -0,0 +1,13 @@
+plugins {
+    id 'java-library'
+}
+
+tasks.withType(JavaCompile).configureEach {
+    options.release = 17
+}
+
+dependencies {
+    api project(':consensusj-jsonrpc')
+
+    testImplementation "org.slf4j:slf4j-jdk14:${slf4jVersion}"       // Runtime implementation of slf4j
+}

consensusj-jsonrpc-unix/src/main/java/org/consensusj/jsonrpc/unix/JsonRpcClientUnixSocket.java

@@ -0,0 +1,85 @@
+package org.consensusj.jsonrpc.unix;
+
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.consensusj.jsonrpc.DefaultRpcClient;
+import org.consensusj.jsonrpc.JsonRpcMessage;
+import org.consensusj.jsonrpc.JsonRpcRequest;
+import org.consensusj.jsonrpc.JsonRpcResponse;
+import org.consensusj.jsonrpc.JsonRpcStatusException;
+import org.consensusj.jsonrpc.JsonRpcTransport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.StandardProtocolFamily;
+import java.net.URI;
+import java.net.UnixDomainSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Path;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Proof-of-concept UNIX domain socket JsonRpc Client (works with {@link UnixSocketEchoServer} and {@code lightningd}.)
+ */
+public class JsonRpcClientUnixSocket implements JsonRpcTransport<JavaType> {
+    private static final Logger log = LoggerFactory.getLogger(JsonRpcClientUnixSocket.class);
+    private final UnixDomainSocketAddress socketAddress;
+    private final JsonRpcUnixSocketMapper socketMapper;
+
+    public static void main(String[] args) throws IOException {
+        boolean useEchoServer = true;
+        Path socketPath = useEchoServer ? UnixSocketEchoServer.getTestPath() : getLightningRpcPath();
+        UnixSocketTransportFactory factory  = new UnixSocketTransportFactory(socketPath);
+        try (var client = new DefaultRpcClient(factory, JsonRpcMessage.Version.V2)) {
+            JsonNode response = client.send("getinfo", JsonNode.class);
+            System.out.println(response);
+        }
+    }
+
+    public JsonRpcClientUnixSocket(Path socketPath, ObjectMapper mapper) {
+        socketAddress = UnixDomainSocketAddress.of(socketPath);
+        socketMapper = new JsonRpcUnixSocketMapper(mapper);
+
+        // TODO: Delete on close:
+        // Files.deleteIfExists(socketPath);
+    }
+
+    @Override
+    public <R> JsonRpcResponse<R> sendRequestForResponse(JsonRpcRequest request, JavaType responseType) throws IOException, JsonRpcStatusException {
+        SocketChannel channel = SocketChannel.open(StandardProtocolFamily.UNIX);
+        // TODO: Use Selectable channel for async??
+        channel.connect(socketAddress);
+        ByteBuffer buffer = socketMapper.serializeRequest(request);
+        while (buffer.hasRemaining()) {
+            channel.write(buffer);
+        }
+        // TODO: Read response see https://www.baeldung.com/java-unix-domain-socket
+        // See also: https://nipafx.dev/java-unix-domain-sockets/
+        // And: https://www.linkedin.com/pulse/java-sockets-io-blocking-non-blocking-asynchronous-aliaksandr-liakh/
+        JsonRpcResponse<R> responseJson = null;
+        try {
+            responseJson = socketMapper.readSocketResponse(request, responseType, channel);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        channel.close();
+        return responseJson;
+    }
+
+    @Override
+    public <R> CompletableFuture<JsonRpcResponse<R>> sendRequestForResponseAsync(JsonRpcRequest request, JavaType responseType) {
+        return supplyAsync(() -> this.sendRequestForResponse(request, responseType));
+    }
+
+    @Override
+    public URI getServerURI() {
+        return socketAddress.getPath().toUri();
+    }
+
+    public static Path getLightningRpcPath() {
+        return Path.of(System.getProperty("user.home")).resolve(".lightning/regtest/lightning-rpc");
+    }
+}

consensusj-jsonrpc-unix/src/main/java/org/consensusj/jsonrpc/unix/JsonRpcUnixSocketMapper.java

@@ -0,0 +1,98 @@
+package org.consensusj.jsonrpc.unix;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.consensusj.jsonrpc.JsonRpcRequest;
+import org.consensusj.jsonrpc.JsonRpcResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.Optional;
+
+/**
+ * Common code for JsonRpc UNIX Sockets clients and servers
+ */
+public class JsonRpcUnixSocketMapper {
+    private static final Logger log = LoggerFactory.getLogger(JsonRpcUnixSocketMapper.class);
+
+    private final ObjectMapper mapper;
+
+    /**
+     * @param mapper Jackson mapper
+     */
+    public JsonRpcUnixSocketMapper(ObjectMapper mapper) {
+        this.mapper = mapper;
+    }
+
+    public <R> JsonRpcResponse<R> readSocketResponse(JsonRpcRequest request, JavaType responseType, SocketChannel channel) throws IOException, InterruptedException {
+        Optional<String> resp = Optional.empty();
+        while (resp.isEmpty()) {
+            Thread.sleep(100);
+            resp = readSocketMessage(channel);
+            resp.ifPresent(System.out::println);
+        }
+        JsonRpcResponse<R> responseJson = deserializeResponse(responseType, resp.orElseThrow());
+        return responseJson;
+    }
+
+    public Optional<String> readSocketMessage(SocketChannel channel) throws IOException {
+        ByteBuffer buffer = ByteBuffer.allocate(10240);
+        int bytesRead = channel.read(buffer);
+        if (bytesRead < 0)
+            return Optional.empty();
+
+        byte[] bytes = new byte[bytesRead];
+        buffer.flip();
+        buffer.get(bytes);
+        String message = new String(bytes);
+        return Optional.of(message);
+    }
+
+    public <R> JsonRpcResponse<R> deserializeResponse(JavaType responseType, String s) throws JsonProcessingException {
+        JsonRpcResponse<R> responseJson;
+        log.debug("Response String: {}", s);
+        try {
+            responseJson = mapper.readValue(s, responseType);
+        } catch (JsonProcessingException e) {
+            log.error("JsonProcessingException: ", e);
+            // TODO: Map to some kind of JsonRPC exception similar to JsonRPCStatusException
+            throw e;
+        }
+        return responseJson;
+    }
+
+    public JsonRpcRequest deserializeRequest(String s) throws JsonProcessingException {
+        JsonRpcRequest requestJson;
+        log.debug("Request String: {}", s);
+        try {
+            requestJson = mapper.readValue(s, JsonRpcRequest.class);
+        } catch (JsonProcessingException e) {
+            log.error("JsonProcessingException: ", e);
+            // TODO: Map to some kind of JsonRPC exception similar to JsonRPCStatusException
+            throw e;
+        }
+        return requestJson;
+    }
+
+    public ByteBuffer serializeRequest(JsonRpcRequest request) throws JsonProcessingException {
+        String message = mapper.writeValueAsString(request);
+        ByteBuffer buffer = ByteBuffer.allocate(1024);
+        buffer.clear();
+        buffer.put(message.getBytes());
+        buffer.flip();
+        return buffer;
+    }
+
+    public <R> ByteBuffer serializeResponse(JsonRpcResponse<R> response) throws JsonProcessingException {
+        String message = mapper.writeValueAsString(response);
+        ByteBuffer buffer = ByteBuffer.allocate(1024);
+        buffer.clear();
+        buffer.put(message.getBytes());
+        buffer.flip();
+        return buffer;
+    }
+}

consensusj-jsonrpc-unix/src/main/java/org/consensusj/jsonrpc/unix/UnixSocketEchoServer.java

@@ -0,0 +1,89 @@
+package org.consensusj.jsonrpc.unix;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.consensusj.jsonrpc.JsonRpcError;
+import org.consensusj.jsonrpc.JsonRpcRequest;
+import org.consensusj.jsonrpc.JsonRpcResponse;
+
+import java.io.IOException;
+import java.net.StandardProtocolFamily;
+import java.net.UnixDomainSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ *  For possible implementation with CompletableFuture,
+ *  See <a href="https://github.com/IBM/java-async-util/blob/master/asyncutil/src/test/java/com/ibm/asyncutil/examples/nio/nio.md#nio-bridge">...</a>
+ *
+ */
+public class UnixSocketEchoServer {
+    private final UnixDomainSocketAddress socketAddress;
+    private final JsonRpcUnixSocketMapper socketMapper;
+
+    public static void main(String[] args) throws IOException, InterruptedException {
+        Path socketPath = getTestPath();
+        Files.deleteIfExists(socketPath);
+        UnixSocketEchoServer server = new UnixSocketEchoServer(socketPath);
+        server.run();
+    }
+
+    public UnixSocketEchoServer(Path socketPath) {
+        socketAddress = UnixDomainSocketAddress.of(socketPath);
+        socketMapper = new JsonRpcUnixSocketMapper(getMapper());
+
+        // TODO: Delete on close:
+        // Files.deleteIfExists(socketPath);
+    }
+
+    public void run() throws IOException, InterruptedException {
+        ServerSocketChannel serverChannel = ServerSocketChannel.open(StandardProtocolFamily.UNIX);
+        serverChannel.bind(socketAddress);
+        SocketChannel channel = serverChannel.accept();
+        while (true) {
+            var optMessage = socketMapper.readSocketMessage(channel);
+            if (optMessage.isPresent()) {
+                processMessage(channel, optMessage.get());
+            }
+            channel.close();
+            channel = serverChannel.accept();
+            Thread.sleep(100);
+        }
+    }
+
+    private void processMessage(SocketChannel channel, String message) throws IOException {
+        System.out.printf("[Client message] %s\n", message);
+        JsonRpcRequest request;
+        try {
+            request = socketMapper.deserializeRequest(message);
+            System.out.println("Got " + request.getMethod() + " request");
+            JsonRpcResponse<?> response = switch (request.getMethod()) {
+                case "getinfo" -> new JsonRpcResponse<>(request, "Echo GETINFO Response");
+                case "help" -> new JsonRpcResponse<>(request, "echo message (TBD)\ngetinfo\nhelp\nstop (TBD)\n");
+                default -> new JsonRpcResponse<>(request, JsonRpcError.of(JsonRpcError.Error.METHOD_NOT_FOUND));
+            };
+            ByteBuffer buffer = socketMapper.serializeResponse(response);
+            while (buffer.hasRemaining()) {
+                channel.write(buffer);
+            }
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    static ObjectMapper getMapper() {
+        var mapper = new ObjectMapper();
+        // TODO: Provide external API to configure FAIL_ON_UNKNOWN_PROPERTIES
+        // TODO: Remove "ignore unknown" annotations on various POJOs that we've defined.
+        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        return mapper;
+    }
+
+    static Path getTestPath() {
+        return Path.of(System.getProperty("user.home")).resolve("consensusj.socket");
+    }
+}

consensusj-jsonrpc-unix/src/main/java/org/consensusj/jsonrpc/unix/UnixSocketTransportFactory.java

@@ -0,0 +1,29 @@
+package org.consensusj.jsonrpc.unix;
+
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.consensusj.jsonrpc.DefaultRpcClient;
+import org.consensusj.jsonrpc.JsonRpcTransport;
+
+import java.nio.file.Path;
+
+/**
+ * Factory for creating {@link JsonRpcClientUnixSocket} from an {@link ObjectMapper}. The
+ * factory instance is created with the {@link Path} to the desired socket.
+ */
+public class UnixSocketTransportFactory implements DefaultRpcClient.TransportFactory {
+    private final Path socketPath;
+
+    /**
+     * Create a factory instance for a given Unix socket path.
+     * @param socketPath Path to the desired socket.
+     */
+    public UnixSocketTransportFactory(Path socketPath) {
+        this.socketPath = socketPath;
+    }
+
+    @Override
+    public JsonRpcTransport<JavaType> create(ObjectMapper mapper) {
+        return new JsonRpcClientUnixSocket(socketPath, mapper);
+    }
+}

settings.gradle

@@ -29,6 +29,7 @@ if (JavaVersion.current().compareTo(JavaVersion.VERSION_21) < 0) {
 
 // JDK 17
     include 'consensusj-jrpc-echod'             // JSON-RPC echo server daemon
+    include 'consensusj-jsonrpc-unix'           // UNIX Domain Sockets for JSONRPC
     include 'cj-bitcoinj-spock'                 // Spock tests/demos of basic bitcoinj capabilities
     include 'cj-bitcoinj-dsl-js'                // JavaScript DSL for bitcoinj via Nashorn
     include 'cj-btc-walletd'                    // SPV Wallet Daemon using bitcoinj WalletAppKit