Implement a more robust IPC system between the launcher and client

apps/app/src/api/oauth_utils/auth_code_reply.rs

@@ -11,18 +11,16 @@
 //! [RFC 8252]: https://datatracker.ietf.org/doc/html/rfc8252
 
 use std::{
-    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
+    net::SocketAddr,
     sync::{LazyLock, Mutex},
     time::Duration,
 };
 
 use hyper::body::Incoming;
 use hyper_util::rt::{TokioIo, TokioTimer};
 use theseus::ErrorKind;
-use tokio::{
-    net::TcpListener,
-    sync::{broadcast, oneshot},
-};
+use theseus::prelude::tcp_listen_any_loopback;
+use tokio::sync::{broadcast, oneshot};
 
 static SERVER_SHUTDOWN: LazyLock<broadcast::Sender<()>> =
     LazyLock::new(|| broadcast::channel(1024).0);
@@ -35,17 +33,7 @@ static SERVER_SHUTDOWN: LazyLock<broadcast::Sender<()>> =
 pub async fn listen(
     listen_socket_tx: oneshot::Sender<Result<SocketAddr, theseus::Error>>,
 ) -> Result<Option<String>, theseus::Error> {
-    // IPv4 is tried first for the best compatibility and performance with most systems.
-    // IPv6 is also tried in case IPv4 is not available. Resolving "localhost" is avoided
-    // to prevent failures deriving from improper name resolution setup. Any available
-    // ephemeral port is used to prevent conflicts with other services. This is all as per
-    // RFC 8252's recommendations
-    const ANY_LOOPBACK_SOCKET: &[SocketAddr] = &[
-        SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0),
-        SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0),
-    ];
-
-    let listener = match TcpListener::bind(ANY_LOOPBACK_SOCKET).await {
+    let listener = match tcp_listen_any_loopback().await {
         Ok(listener) => {
             listen_socket_tx
                 .send(listener.local_addr().map_err(|e| {

packages/app-lib/build.rs

@@ -53,7 +53,6 @@ fn build_java_jars() {
         .arg("build")
         .arg("--no-daemon")
         .arg("--console=rich")
-        .arg("--info")
         .current_dir(dunce::canonicalize("java").unwrap())
         .status()
         .expect("Failed to wait on Gradle build");

packages/app-lib/java/build.gradle.kts

@@ -11,6 +11,7 @@ repositories {
 dependencies {
     implementation("org.ow2.asm:asm:9.8")
     implementation("org.ow2.asm:asm-tree:9.8")
+    implementation("com.google.code.gson:gson:2.13.1")
 
     testImplementation(libs.junit.jupiter)
     testRuntimeOnly("org.junit.platform:junit-platform-launcher")

packages/app-lib/java/src/main/java/com/modrinth/theseus/MinecraftLaunch.java

@@ -1,55 +1,31 @@
 package com.modrinth.theseus;
 
-import java.io.ByteArrayOutputStream;
+import com.modrinth.theseus.rpc.RpcHandlers;
+import com.modrinth.theseus.rpc.TheseusRpc;
 import java.io.IOException;
 import java.lang.reflect.AccessibleObject;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
 
 public final class MinecraftLaunch {
     public static void main(String[] args) throws IOException, ReflectiveOperationException {
         final String mainClass = args[0];
         final String[] gameArgs = Arrays.copyOfRange(args, 1, args.length);
 
         System.setProperty("modrinth.process.args", String.join("\u001f", gameArgs));
-        parseInput();
 
-        relaunch(mainClass, gameArgs);
-    }
-
-    private static void parseInput() throws IOException {
-        final ByteArrayOutputStream line = new ByteArrayOutputStream();
-        while (true) {
-            final int b = System.in.read();
-            if (b < 0) {
-                throw new IllegalStateException("Stdin terminated while parsing");
-            }
-            if (b != '\n') {
-                line.write(b);
-                continue;
-            }
-            if (handleLine(line.toString("UTF-8"))) {
-                break;
-            }
-            line.reset();
-        }
-    }
-
-    private static boolean handleLine(String line) {
-        final String[] parts = line.split("\t", 2);
-        switch (parts[0]) {
-            case "property": {
-                final String[] keyValue = parts[1].split("\t", 2);
-                System.setProperty(keyValue[0], keyValue[1]);
-                return false;
-            }
-            case "launch":
-                return true;
-        }
+        final CompletableFuture<Void> waitForLaunch = new CompletableFuture<>();
+        TheseusRpc.connectAndStart(
+                System.getProperty("modrinth.internal.ipc.host"),
+                Integer.getInteger("modrinth.internal.ipc.port"),
+                new RpcHandlers()
+                        .handler("set_system_property", String.class, String.class, System::setProperty)
+                        .handler("launch", () -> waitForLaunch.complete(null)));
 
-        System.err.println("Unknown input line " + line);
-        return false;
+        waitForLaunch.join();
+        relaunch(mainClass, gameArgs);
     }
 
     private static void relaunch(String mainClassName, String[] args) throws ReflectiveOperationException {

packages/app-lib/java/src/main/java/com/modrinth/theseus/rpc/RpcHandlers.java

@@ -0,0 +1,46 @@
+package com.modrinth.theseus.rpc;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonNull;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+public class RpcHandlers {
+    private final Map<String, Function<JsonElement[], JsonElement>> handlers = new HashMap<>();
+    private boolean frozen;
+
+    public RpcHandlers handler(String functionName, Runnable handler) {
+        return addHandler(functionName, args -> {
+            handler.run();
+            return JsonNull.INSTANCE;
+        });
+    }
+
+    public <A, B> RpcHandlers handler(
+            String functionName, Class<A> arg1Type, Class<B> arg2Type, BiConsumer<A, B> handler) {
+        return addHandler(functionName, args -> {
+            if (args.length != 2) {
+                throw new IllegalArgumentException(functionName + " expected 2 arguments");
+            }
+            final A arg1 = TheseusRpc.GSON.fromJson(args[0], arg1Type);
+            final B arg2 = TheseusRpc.GSON.fromJson(args[1], arg2Type);
+            handler.accept(arg1, arg2);
+            return JsonNull.INSTANCE;
+        });
+    }
+
+    private RpcHandlers addHandler(String functionName, Function<JsonElement[], JsonElement> handler) {
+        if (frozen) {
+            throw new IllegalStateException("Cannot add handler to frozen RpcHandlers instance");
+        }
+        handlers.put(functionName, handler);
+        return this;
+    }
+
+    Map<String, Function<JsonElement[], JsonElement>> build() {
+        frozen = true;
+        return handlers;
+    }
+}

packages/app-lib/java/src/main/java/com/modrinth/theseus/rpc/RpcMethodException.java

@@ -0,0 +1,9 @@
+package com.modrinth.theseus.rpc;
+
+public class RpcMethodException extends RuntimeException {
+    private static final long serialVersionUID = 1922360184188807964L;
+
+    public RpcMethodException(String message) {
+        super(message);
+    }
+}

packages/app-lib/java/src/main/java/com/modrinth/theseus/rpc/TheseusRpc.java

@@ -0,0 +1,183 @@
+package com.modrinth.theseus.rpc;
+
+import com.google.gson.*;
+import com.google.gson.reflect.TypeToken;
+import java.io.*;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+public final class TheseusRpc {
+    static final Gson GSON = new GsonBuilder()
+            .setStrictness(Strictness.STRICT)
+            .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
+            .disableHtmlEscaping()
+            .create();
+    private static final TypeToken<RpcMessage> MESSAGE_TYPE = TypeToken.get(RpcMessage.class);
+
+    private static final AtomicReference<TheseusRpc> RPC = new AtomicReference<>();
+
+    private final BlockingQueue<RpcMessage> mainThreadQueue = new LinkedBlockingQueue<>();
+    private final Map<UUID, ResponseWaiter<?>> awaitingResponse = new ConcurrentHashMap<>();
+    private final Map<String, Function<JsonElement[], JsonElement>> handlers;
+    private final Socket socket;
+
+    private TheseusRpc(Socket socket, RpcHandlers handlers) {
+        this.socket = socket;
+        this.handlers = handlers.build();
+    }
+
+    public static void connectAndStart(String host, int port, RpcHandlers handlers) throws IOException {
+        if (RPC.get() != null) {
+            throw new IllegalStateException("Can only connect to RPC once");
+        }
+
+        final Socket socket = new Socket(host, port);
+        final TheseusRpc rpc = new TheseusRpc(socket, handlers);
+        final Thread mainThread = new Thread(rpc::mainThread, "Theseus RPC Main");
+        final Thread readThread = new Thread(rpc::readThread, "Theseus RPC Read");
+        mainThread.setDaemon(true);
+        readThread.setDaemon(true);
+        mainThread.start();
+        readThread.start();
+        RPC.set(rpc);
+    }
+
+    public static TheseusRpc getRpc() {
+        final TheseusRpc rpc = RPC.get();
+        if (rpc == null) {
+            throw new IllegalStateException("Called getRpc before RPC initialized");
+        }
+        return rpc;
+    }
+
+    public <T> CompletableFuture<T> callMethod(TypeToken<T> returnType, String method, Object... args) {
+        final JsonElement[] jsonArgs = new JsonElement[args.length];
+        for (int i = 0; i < args.length; i++) {
+            jsonArgs[i] = GSON.toJsonTree(args[i]);
+        }
+
+        final RpcMessage message = new RpcMessage(method, jsonArgs);
+        final ResponseWaiter<T> responseWaiter = new ResponseWaiter<>(returnType);
+        awaitingResponse.put(message.id, responseWaiter);
+        mainThreadQueue.add(message);
+        return responseWaiter.future;
+    }
+
+    private void mainThread() {
+        try {
+            final Writer writer = new OutputStreamWriter(socket.getOutputStream(), StandardCharsets.UTF_8);
+            while (true) {
+                final RpcMessage message = mainThreadQueue.take();
+                final RpcMessage toSend;
+                if (message.isForSending) {
+                    toSend = message;
+                } else {
+                    final Function<JsonElement[], JsonElement> handler = handlers.get(message.method);
+                    if (handler == null) {
+                        System.err.println("Unknown theseus RPC method " + message.method);
+                        continue;
+                    }
+                    RpcMessage response;
+                    try {
+                        response = new RpcMessage(message.id, handler.apply(message.args));
+                    } catch (Exception e) {
+                        response = new RpcMessage(message.id, e.toString());
+                    }
+                    toSend = response;
+                }
+                GSON.toJson(toSend, writer);
+                writer.write('\n');
+                writer.flush();
+            }
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        } catch (InterruptedException ignored) {
+        }
+    }
+
+    private void readThread() {
+        try {
+            final BufferedReader reader =
+                    new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
+            while (true) {
+                final RpcMessage message = GSON.fromJson(reader.readLine(), MESSAGE_TYPE);
+                if (message.method == null) {
+                    final ResponseWaiter<?> waiter = awaitingResponse.get(message.id);
+                    if (waiter != null) {
+                        handleResponse(waiter, message);
+                    }
+                } else {
+                    mainThreadQueue.put(message);
+                }
+            }
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        } catch (InterruptedException ignored) {
+        }
+    }
+
+    private <T> void handleResponse(ResponseWaiter<T> waiter, RpcMessage message) {
+        if (message.error != null) {
+            waiter.future.completeExceptionally(new RpcMethodException(message.error));
+            return;
+        }
+        try {
+            waiter.future.complete(GSON.fromJson(message.response, waiter.type));
+        } catch (JsonSyntaxException e) {
+            waiter.future.completeExceptionally(e);
+        }
+    }
+
+    private static class RpcMessage {
+        final UUID id;
+        final String method; // Optional
+        final JsonElement[] args; // Optional
+        final JsonElement response; // Optional
+        final String error; // Optional
+        final transient boolean isForSending;
+
+        RpcMessage(String method, JsonElement[] args) {
+            id = UUID.randomUUID();
+            this.method = method;
+            this.args = args;
+            response = null;
+            error = null;
+            isForSending = true;
+        }
+
+        RpcMessage(UUID id, JsonElement response) {
+            this.id = id;
+            method = null;
+            args = null;
+            this.response = response;
+            error = null;
+            isForSending = true;
+        }
+
+        RpcMessage(UUID id, String error) {
+            this.id = id;
+            method = null;
+            args = null;
+            response = null;
+            this.error = error;
+            isForSending = true;
+        }
+    }
+
+    private static class ResponseWaiter<T> {
+        final TypeToken<T> type;
+        final CompletableFuture<T> future = new CompletableFuture<>();
+
+        ResponseWaiter(TypeToken<T> type) {
+            this.type = type;
+        }
+    }
+}

packages/app-lib/src/api/mod.rs

@@ -35,6 +35,9 @@ pub mod prelude {
         jre, metadata, minecraft_auth, mr_auth, pack, process,
         profile::{self, Profile, create},
         settings,
-        util::io::{IOError, canonicalize},
+        util::{
+            io::{IOError, canonicalize},
+            network::tcp_listen_any_loopback,
+        },
     };
 }

packages/app-lib/src/error.rs

@@ -151,6 +151,9 @@ pub enum ErrorKind {
         "A skin texture must have a dimension of either 64x64 or 64x32 pixels"
     )]
     InvalidSkinTexture,
+
+    #[error("RPC error: {0}")]
+    RpcError(String),
 }
 
 #[derive(Debug)]