reworked MessageBroker, now also with in-memory implementation

core-bukkit-impl/src/main/java/de/kentoj/scrow/bukkit/ScrowAPISurface.java

@@ -12,7 +12,8 @@ import de.kentoj.scrowlib.friends.friendship.PostgresFriendshipService;
 import de.kentoj.scrowlib.friends.request.InMemoryFriendRequestService;
 import de.kentoj.scrowlib.friends.request.PostgresFriendRequestService;
 import de.kentoj.scrowlib.instancemanager.InstanceManagerImpl;
-import de.kentoj.scrowlib.messaging.MessageBroker;
+import de.kentoj.scrowlib.messaging.InMemoyMessageBroker;
+import de.kentoj.scrowlib.messaging.NatsMessageBroker;
 import de.kentoj.scrowlib.player.NetworkPlayerImpl;
 import de.kentoj.scrowlib.utils.EnvUtils;
 import io.nats.client.Nats;
@@ -33,25 +34,33 @@ public class ScrowAPISurface {
     public static void initScrowAPI(Plugin plugin) {
         ScrowAPI.setCorePlugin(plugin);
 
+        var natsHost = System.getenv("HOST_NATS");
+        if (natsHost!= null) {
             try {
-            String natsHost = EnvUtils.envOrThrow("HOST_NATS");
                 var options = Options.builder()
                         .server(natsHost)
                         .pedantic()
                         .build();
-            ScrowAPI.setMessageBroker(new MessageBroker(Nats.connect(options)));
+                ScrowAPI.setMessageBroker(new NatsMessageBroker(Nats.connect(options)));
             } catch (IOException | InterruptedException e) {
                 throw new RuntimeException(e);
             }
+        } else {
+            log.error("HOST_NATS not set -> Using in-memory dummy message broker");
+            ScrowAPI.setMessageBroker(new InMemoyMessageBroker());
+        }
 
+        var hostPostgres = System.getenv().get("HOST_POSTGRES");
+        if (hostPostgres != null) {
             try {
                 Class.forName("org.postgresql.Driver", true, ScrowAPISurface.class.getClassLoader());
                 var props = new Properties();
                 props.setProperty("user", "postgres");
-            ScrowAPI.setJdbi(Jdbi.create(EnvUtils.envOrThrow("HOST_POSTGRES"), props));
+                ScrowAPI.setJdbi(Jdbi.create(hostPostgres));
             } catch (ClassNotFoundException e) {
                 throw new RuntimeException(e);
-        } catch (IllegalStateException e) {
+            }
+        } else {
             log.error("HOST_POSTGRES not set -> Using in-memory database");
         }
 

core-lib/src/main/java/de/kentoj/scrowlib/instancemanager/InstanceManagerImpl.java

@@ -1,14 +1,13 @@
 package de.kentoj.scrowlib.instancemanager;
 
 import de.kentoj.scrowlib.messaging.MessageBroker;
+import de.kentoj.scrowlib.messaging.RemoteException;
 import lombok.RequiredArgsConstructor;
 import org.bson.Document;
 import org.jspecify.annotations.Nullable;
 
 import java.util.List;
 
-import static de.kentoj.scrowlib.utils.ResultUtils.unwrap;
-
 @RequiredArgsConstructor
 public class InstanceManagerImpl implements InstanceManager {
 
@@ -16,31 +15,34 @@ public class InstanceManagerImpl implements InstanceManager {
 
     @Override
     public ServerInstance deployInstance(String template) {
-        var result = messageBroker.request("instance.deploy", new Document("template", template));
+        var document = messageBroker.request("instance.deploy", new Document("template", template));
-        var document = unwrap(result, "failed to deploy instance of " + template);
         return ServerInstance.fromDocument(document);
     }
 
     @Override
     public boolean destroyInstance(String handle) {
-        var result = messageBroker.request("instance.destroy", new Document("handle", handle));
+        try {
-        return unwrap(result.mapSuccess(__ -> true)
+            messageBroker.request("instance.destroy", new Document("handle", handle));
-                .recover(this::wasNotFound, __ -> false), "failed to destroy instance " + handle);
+            return true;
+        } catch (RemoteException ex) {
+            if (wasNotFound(ex.getMessage())) return false;
+            throw ex;
+        }
     }
 
     @Override
     public List<ServerInstance> getInstances() {
-        var result = messageBroker.request("instance.list", new Document());
+        var document = messageBroker.request("instance.list", new Document());
-        return unwrap(result.mapSuccess(doc -> doc.getList("instances", Document.class)
+        return document.getList("instances", Document.class)
-                        .stream().map(ServerInstance::fromDocument).toList()),
+                .stream()
-                "failed to get instances");
+                .map(ServerInstance::fromDocument)
+                .toList();
     }
 
     @Override
     public @Nullable ServerInstance getInstance(String handle) {
-        var result = messageBroker.request("instance.get", new Document("handle", handle));
+        var document = messageBroker.request("instance.get", new Document("handle", handle));
-        return unwrap(result.mapSuccess(ServerInstance::fromDocument)
+        return ServerInstance.fromDocument(document);
-                .recover(this::wasNotFound, __ -> null), "failed to get instance " + handle);
     }
 
     private boolean wasNotFound(String failure) {

core-lib/src/main/java/de/kentoj/scrowlib/messaging/InMemoyMessageBroker.java

@@ -0,0 +1,52 @@
+package de.kentoj.scrowlib.messaging;
+
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+
+@Slf4j
+public class InMemoyMessageBroker implements MessageBroker {
+
+    private final Map<String, List<Consumer<Document>>> subscribers = new HashMap<>();
+    private final Map<String, RequestHandler> handlers = new HashMap<>();
+
+    @Override
+    public void publish(String subject, Document document) {
+        var subs = subscribers.get(subject);
+        if (subs == null) return;
+        subs.forEach(s -> s.accept(document));
+    }
+
+    @Override
+    public void subscribe(String subject, Consumer<Document> consumer) {
+        subscribers.computeIfAbsent(subject, __ -> new ArrayList<>())
+                .add(consumer);
+    }
+
+    @Override
+    public void handle(String subject, RequestHandler handler) {
+        var oldValue = handlers.put(subject, handler);
+        if (oldValue != null) log.warn("handler overwrites previous handler for subject " + subject);
+    }
+
+    @Override
+    public Document request(String subject, Document document) {
+        var handler = handlers.get(subject);
+        if (handler == null)
+            throw new RuntimeException("No request handler for " + subject + " found");
+        try {
+            return handler.handle(document).get();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        } catch (ExecutionException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

core-lib/src/main/java/de/kentoj/scrowlib/messaging/MessageBroker.java

@@ -1,84 +1,41 @@
 package de.kentoj.scrowlib.messaging;
 
-import com.leakyabstractions.result.api.Result;
-import io.nats.client.Connection;
-import io.nats.client.Dispatcher;
-import lombok.extern.slf4j.Slf4j;
 import org.bson.Document;
 
-import java.util.Optional;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.function.Consumer;
 
-@Slf4j
+public interface MessageBroker {
-public class MessageBroker {
-
-    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
-    private final Connection nats;
-    protected final Dispatcher dispatcher;
-
-    public MessageBroker(Connection nats) {
-        this.nats = nats;
-        this.dispatcher = nats.createDispatcher();
-    }
 
     /**
      * Runs blocking.
      * publishes to the message broker
+     *
      * @param subject <a href="https://docs.nats.io/nats-concepts/subjects">subject</a> to publish to
      */
-    public void publish(String subject, Document document) {
+    void publish(String subject, Document document);
-        nats.publish(subject, DocumentRepository.toBytes(document));
+
-    }
+    /**
+     * subscribes to a subject, consumer may run asynchronously
+     *
+     * @param subject <a href="https://docs.nats.io/nats-concepts/subjects">subject</a> to publish to
+     */
+    void subscribe(String subject, Consumer<Document> consumer);
+
+    /**
+     * Registers a handler for a subject, the handler will receive requests and compute some result.
+     * The result is sent back as a reply.
+     *
+     * @param subject <a href="https://docs.nats.io/nats-concepts/subjects">subject</a> to publish to
+     */
+    void handle(String subject, RequestHandler handler);
 
     /**
      * Runs blocking.
-     * publishes to the message broker
+     * Executes an RPC call: The document is published and this method starts listening for incoming
-     * @param subject <a href="https://docs.nats.io/nats-concepts/subjects">subject</a> to publish to
-     */
-    public void publish(String subject, Result<Document, String> result) {
-        this.publish(subject, ResultRepository.toDocument(result));
-    }
-
-    /**
-     * subscribes to a subject, handler may run asynchronously
-     */
-    public void subscribe(String subject, Consumer<Document> handler) {
-        dispatcher.subscribe(subject, msg ->
-                executor.submit(() -> handler.accept(DocumentRepository.fromMessage(msg))));
-    }
-
-    /**
-     * handler may run asynchronously
-     */
-    public void handle(String subject, RequestHandler handler) {
-        dispatcher.subscribe(subject, msg -> {
-            if (msg.getReplyTo() == null) {
-                log.warn("request in {} has replyTo unset; ignoring", subject);
-                return;
-            }
-            handler.handle(DocumentRepository.fromMessage(msg))
-                    .thenAccept(result ->
-                            executor.submit(() -> publish(msg.getReplyTo(), result))
-                    );
-        });
-    }
-
-    /**
-     * Runs blocking.
-     * Executes a RPC call: The document is published and this method starts listening for incoming
      * replies. The returned Result holds either a document with the requested data, or an error as a string
+     *
+     * @param subject <a href="https://docs.nats.io/nats-concepts/subjects">subject</a> to publish to
+     * @return RemoteException if the remote throws an exception while handling the request
      */
-    public Result<Document, String> request(String subject, Document document) {
+    Document request(String subject, Document document);
-        var response = nats.request(subject, DocumentRepository.toBytes(document)).join();
-        return ResultRepository.fromDocument(DocumentRepository.fromMessage(response));
-    }
-
-    public static <S> S unwrap(Result<S, String> result, String message) {
-        Optional<String> err = result.getFailure();
-        if (err.isPresent())
-            throw new RuntimeException(message + ": " + err.orElseThrow());
-        return result.getSuccess().orElseThrow();
-    }
 }

core-lib/src/main/java/de/kentoj/scrowlib/messaging/NatsMessageBroker.java

@@ -0,0 +1,78 @@
+package de.kentoj.scrowlib.messaging;
+
+import com.leakyabstractions.result.api.Result;
+import com.leakyabstractions.result.core.Results;
+import de.kentoj.scrowlib.utils.DocumentRepository;
+import io.nats.client.Connection;
+import io.nats.client.Dispatcher;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+
+@Slf4j
+public class NatsMessageBroker implements MessageBroker {
+
+    private static final ExecutorService EXECUTOR = Executors.newVirtualThreadPerTaskExecutor();
+    private final Connection nats;
+    private final Dispatcher dispatcher;
+
+    public NatsMessageBroker(Connection nats) {
+        this.nats = nats;
+        this.dispatcher = nats.createDispatcher();
+    }
+
+    @Override
+    public void publish(String subject, Document document) {
+        nats.publish(subject, DocumentRepository.toBytes(document));
+    }
+
+    @Override
+    public void subscribe(String subject, Consumer<Document> consumer) {
+        dispatcher.subscribe(subject, msg ->
+                EXECUTOR.submit(() -> consumer.accept(DocumentRepository.fromBytes(msg.getData()))));
+    }
+
+    @Override
+    public void handle(String subject, RequestHandler handler) {
+        dispatcher.subscribe(subject, msg -> {
+            if (msg.getReplyTo() == null) {
+                log.warn("request in {} has replyTo unset; ignoring", subject);
+                return;
+            }
+
+            Result<Document, String> result;
+            try {
+                var reply = handler.handle(DocumentRepository.fromBytes(msg.getData())).get();
+                result = Results.success(reply);
+            } catch (ExecutionException ex) {
+                result = Results.failure(ex.getCause().getMessage());
+            } catch (InterruptedException ex) {
+                Thread.currentThread().interrupt();
+                throw ex;
+            }
+
+            final var finalResult = result;
+            EXECUTOR.submit(() -> publish(msg.getReplyTo(), ResultRepository.toDocument(finalResult)));
+        });
+    }
+
+    @Override
+    public Document request(String subject, Document document) {
+        var reply = nats.request(subject, DocumentRepository.toBytes(document)).join();
+        var result = ResultRepository.fromDocument(DocumentRepository.fromBytes(reply.getData()));
+        return unwrap(result);
+    }
+
+    @SuppressWarnings("OptionalGetWithoutIsPresent")
+    private static <S> S unwrap(Result<S, String> result) {
+        Optional<String> err = result.getFailure();
+        if (err.isPresent())
+            throw new RemoteException(err.get());
+        return result.getSuccess().get();
+    }
+}

core-lib/src/main/java/de/kentoj/scrowlib/messaging/RequestHandler.java

@@ -1,16 +1,15 @@
 package de.kentoj.scrowlib.messaging;
 
-import com.leakyabstractions.result.api.Result;
 import org.bson.Document;
 
-import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Future;
 
 @FunctionalInterface
 public interface RequestHandler {
     /**
      * Processes a request and replies
      *
-     * @return result to send back to the requesting client.
+     * @return document to reply with
      */
-    CompletionStage<Result<Document, String>> handle(Document doc);
+    Future<Document> handle(Document doc);
 }

core-lib/src/main/java/de/kentoj/scrowlib/messaging/ResultRepository.java

@@ -8,15 +8,15 @@ import org.bson.Document;
 
 
 @NoArgsConstructor(access = AccessLevel.NONE)
-public class ResultRepository {
+class ResultRepository {
 
-    public static Document toDocument(Result<Document, String> result) {
+    static Document toDocument(Result<Document, String> result) {
         return result.hasSuccess()
                 ? new Document("data", result.getSuccess().orElseThrow())
                 : new Document("error", result.getFailure().orElseThrow());
     }
 
-    public static Result<Document, String> fromDocument(Document doc) {
+    static Result<Document, String> fromDocument(Document doc) {
         var error = doc.getString("error");
         return error != null
                 ? Results.failure(error)

core-lib/src/main/java/de/kentoj/scrowlib/messaging/SyncRequestHandler.java

@@ -1,17 +1,16 @@
 package de.kentoj.scrowlib.messaging;
 
-import com.leakyabstractions.result.api.Result;
 import org.bson.Document;
 
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Future;
 
 @FunctionalInterface
-public interface SyncRequestHandler extends RequestHandler{
+public interface SyncRequestHandler extends RequestHandler {
-    Result<Document, String> handleSync(Document doc);
+    Document handleSync(Document doc);
 
     @Override
-    default CompletionStage<Result<Document, String>> handle(Document doc) {
+    default Future<Document> handle(Document doc) {
         return CompletableFuture.completedFuture(handleSync(doc));
     }
 }

core-lib/src/main/java/de/kentoj/scrowlib/player/NetworkPlayer.java

@@ -1,5 +1,6 @@
 package de.kentoj.scrowlib.player;
 
+import de.kentoj.scrowlib.messaging.MessageBroker;
 import net.kyori.adventure.audience.Audience;
 
 /**
@@ -7,15 +8,15 @@ import net.kyori.adventure.audience.Audience;
  * NetworkPlayer represents a Player that MAY be online on any of the minecraft, wrapped
  * in an {@link Audience}
  * servers in the network.
- *
+ * <p>
  * If the player is not online, these methods DO NOTHING.
  * Most methods here do nothing even if the Player is valid(as is the nature of Audiences)
- * 
+ * <p>
- * this interface should be used mainly for basic communication(e.g sending messages).
+ * this interface should be used mainly for basic communication(e.g. sending messages).
  * For more complex operations, use the {@link MessageBroker}.
  *
- * @see {@link NetworkPlayerImpl}
+ * @see NetworkPlayerImpl
- * @see {@link MessageBroker}
+ * @see MessageBroker
  */
 public interface NetworkPlayer extends Audience {
 

core-lib/src/main/java/de/kentoj/scrowlib/utils/DocumentRepository.java

@@ -1,6 +1,5 @@
-package de.kentoj.scrowlib.messaging;
+package de.kentoj.scrowlib.utils;
 
-import io.nats.client.Message;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
 import org.bson.Document;
@@ -10,8 +9,8 @@ import org.jetbrains.annotations.ApiStatus;
 public class DocumentRepository {
 
     @ApiStatus.Obsolete
-    public static Document fromMessage(Message message) {
+    public static Document fromBytes(byte[] bytes) {
-        return Document.parse(new String(message.getData()));
+        return Document.parse(new String(bytes));
     }
 
     @ApiStatus.Obsolete

core-lib/src/main/java/de/kentoj/scrowlib/utils/ResultUtils.java

@@ -1,11 +0,0 @@
-package de.kentoj.scrowlib.utils;
-
-import com.leakyabstractions.result.api.Result;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-
-import java.util.Optional;
-
-@NoArgsConstructor(access = AccessLevel.NONE)
-public class ResultUtils {
-}

core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/CoreVelocityPlugin.java

@@ -26,7 +26,7 @@ import net.kyori.adventure.text.format.NamedTextColor;
 public class CoreVelocityPlugin {
     @Inject
     public CoreVelocityPlugin(ProxyServer server) {
-        ScrowAPISurface.initScrowAPI(server, true);
+        ScrowAPISurface.initScrowAPI(server);
 
         var cmds = ScrowAPI.getCommandApi();
         {

core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/ScrowAPISurface.java

@@ -4,42 +4,53 @@ import com.velocitypowered.api.proxy.ProxyServer;
 import de.kentoj.kencommandapi.CommandAPI;
 import de.kentoj.scrow.velocity.ScrowAPI;
 import de.kentoj.scrowlib.instancemanager.InstanceManagerImpl;
-import de.kentoj.scrowlib.messaging.MessageBroker;
+import de.kentoj.scrowlib.messaging.InMemoyMessageBroker;
+import de.kentoj.scrowlib.messaging.NatsMessageBroker;
 import de.kentoj.scrowlib.player.NetworkPlayerImpl;
-import de.kentoj.scrowlib.utils.EnvUtils;
 import io.nats.client.Nats;
 import io.nats.client.Options;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.core.Jdbi;
 
 import java.io.IOException;
 import java.util.Properties;
 
+@Slf4j
 @NoArgsConstructor(access = AccessLevel.NONE)
 public class ScrowAPISurface {
 
-    public static void initScrowAPI(ProxyServer server, boolean enableDatabase) {
+    public static void initScrowAPI(ProxyServer server) {
+
+        var natsHost = System.getenv("HOST_NATS");
+        if (natsHost != null) {
             try {
-            String natsHost = EnvUtils.envOrThrow("HOST_NATS");
                 var options = Options.builder()
                         .server(natsHost)
                         .pedantic()
                         .build();
-            ScrowAPI.setMessageBroker(new MessageBroker(Nats.connect(options)));
+                ScrowAPI.setMessageBroker(new NatsMessageBroker(Nats.connect(options)));
             } catch (IOException | InterruptedException e) {
                 throw new RuntimeException(e);
             }
+        } else {
+            log.error("HOST_NATS not set -> Using in-memory dummy message broker");
+            ScrowAPI.setMessageBroker(new InMemoyMessageBroker());
+        }
 
-        if (enableDatabase) {
+        var hostPostgres = System.getenv().get("HOST_POSTGRES");
+        if (hostPostgres != null) {
             try {
                 Class.forName("org.postgresql.Driver", true, ScrowAPISurface.class.getClassLoader());
+                var props = new Properties();
+                props.setProperty("user", "postgres");
+                ScrowAPI.setJdbi(Jdbi.create(hostPostgres));
             } catch (ClassNotFoundException e) {
                 throw new RuntimeException(e);
             }
-            var props = new Properties();
+        } else {
-            props.setProperty("user", "postgres");
+            log.error("HOST_POSTGRES not set -> Using in-memory database");
-            ScrowAPI.setJdbi(Jdbi.create(EnvUtils.envOrThrow("HOST_POSTGRES"), props));
         }
 
         ScrowAPI.setInstanceManager(new InstanceManagerImpl(ScrowAPI.getMessageBroker()));

core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/network/SendPlayerWatchdog.java

@@ -1,7 +1,5 @@
 package de.kentoj.scrow.corevelocity.network;
 
-import com.leakyabstractions.result.api.Result;
-import com.leakyabstractions.result.core.Results;
 import com.velocitypowered.api.proxy.ProxyServer;
 import de.kentoj.scrow.velocity.ScrowAPI;
 import de.kentoj.scrowlib.messaging.MessageBroker;
@@ -10,14 +8,16 @@ import lombok.extern.slf4j.Slf4j;
 import org.bson.Document;
 
 import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
-import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executors;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 @Slf4j
 @RequiredArgsConstructor
 public class SendPlayerWatchdog {
 
+    private final ExecutorService EXECUTOR = Executors.newVirtualThreadPerTaskExecutor();
+
     private final MessageBroker messageBroker = ScrowAPI.getMessageBroker();
     private final ProxyServer server;
 
@@ -28,17 +28,14 @@ public class SendPlayerWatchdog {
         ));
     }
 
-    private CompletionStage<Result<Document, String>> sendPlayer(UUID playerId, String handle) {
+    private Future<Document> sendPlayer(UUID playerId, String handle) {
-        return CompletableFuture.supplyAsync(() -> {
+        return EXECUTOR.submit(() -> {
             var player = server.getPlayer(playerId).orElseThrow();
             var targetServer = server.getServer(handle).orElseThrow();
-            try {
             var res = player.createConnectionRequest(targetServer).connect().get();
-                return res.isSuccessful() ? Results.success(new Document()) : Results.failure(res.getStatus().name());
+            if (!res.isSuccessful())
-            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(res.getStatus().name());
-                log.error("failed to send player to {}", handle, e);
+            return new Document();
-                return Results.failure(e.getMessage());
-            }
         });
     }
 }

gradle/libs.versions.toml

@@ -12,7 +12,6 @@ adventure = "5.2.0"
 [plugins]
 shadow = { id = "com.gradleup.shadow", version.ref = "shadow" }
 runpaper = { id = "xyz.jpenilla.run-paper", version.ref = "run-paper" }
-scrow-scrowrepository = { id = "scrow.scrow-repository" }
 scrow-basedependencies = { id = "scrow.base-dependencies" }
 scrow-javalibrary = { id = "scrow.java-library" }
 scrow-database = { id = "scrow.database" }