From 7e5bd1bb6df9692030ebdecef9b4065c1faa9bc6 Mon Sep 17 00:00:00 2001 From: kento2 Date: Tue, 7 Jul 2026 15:05:36 +0200 Subject: [PATCH] reworked MessageBroker, now also with in-memory implementation --- .../kentoj/scrow/bukkit/ScrowAPISurface.java | 45 ++++++----- .../instancemanager/InstanceManagerImpl.java | 30 +++---- .../messaging/InMemoyMessageBroker.java | 52 ++++++++++++ .../scrowlib/messaging/MessageBroker.java | 79 +++++-------------- .../scrowlib/messaging/NatsMessageBroker.java | 78 ++++++++++++++++++ .../scrowlib/messaging/RequestHandler.java | 7 +- .../scrowlib/messaging/ResultRepository.java | 6 +- .../messaging/SyncRequestHandler.java | 9 +-- .../kentoj/scrowlib/player/NetworkPlayer.java | 11 +-- .../DocumentRepository.java | 7 +- .../de/kentoj/scrowlib/utils/ResultUtils.java | 11 --- .../corevelocity/CoreVelocityPlugin.java | 2 +- .../scrow/corevelocity/ScrowAPISurface.java | 43 ++++++---- .../network/SendPlayerWatchdog.java | 25 +++--- gradle/libs.versions.toml | 1 - 15 files changed, 249 insertions(+), 157 deletions(-) create mode 100644 core-lib/src/main/java/de/kentoj/scrowlib/messaging/InMemoyMessageBroker.java create mode 100644 core-lib/src/main/java/de/kentoj/scrowlib/messaging/NatsMessageBroker.java rename core-lib/src/main/java/de/kentoj/scrowlib/{messaging => utils}/DocumentRepository.java (66%) delete mode 100644 core-lib/src/main/java/de/kentoj/scrowlib/utils/ResultUtils.java diff --git a/core-bukkit-impl/src/main/java/de/kentoj/scrow/bukkit/ScrowAPISurface.java b/core-bukkit-impl/src/main/java/de/kentoj/scrow/bukkit/ScrowAPISurface.java index a684854..a3afa63 100644 --- a/core-bukkit-impl/src/main/java/de/kentoj/scrow/bukkit/ScrowAPISurface.java +++ b/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); - try { - String natsHost = EnvUtils.envOrThrow("HOST_NATS"); - var options = Options.builder() - .server(natsHost) - .pedantic() - .build(); - ScrowAPI.setMessageBroker(new MessageBroker(Nats.connect(options))); - } catch (IOException | InterruptedException e) { - throw new RuntimeException(e); + var natsHost = System.getenv("HOST_NATS"); + if (natsHost!= null) { + try { + var options = Options.builder() + .server(natsHost) + .pedantic() + .build(); + 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()); } - 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)); - } catch (ClassNotFoundException e) { - throw new RuntimeException(e); - } catch (IllegalStateException e) { + 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); + } + } else { log.error("HOST_POSTGRES not set -> Using in-memory database"); } diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/instancemanager/InstanceManagerImpl.java b/core-lib/src/main/java/de/kentoj/scrowlib/instancemanager/InstanceManagerImpl.java index 96ed135..a390d7e 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/instancemanager/InstanceManagerImpl.java +++ b/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 = unwrap(result, "failed to deploy instance of " + template); + var document = messageBroker.request("instance.deploy", new Document("template", template)); return ServerInstance.fromDocument(document); } @Override public boolean destroyInstance(String handle) { - var result = messageBroker.request("instance.destroy", new Document("handle", handle)); - return unwrap(result.mapSuccess(__ -> true) - .recover(this::wasNotFound, __ -> false), "failed to destroy instance " + handle); + try { + messageBroker.request("instance.destroy", new Document("handle", handle)); + return true; + } catch (RemoteException ex) { + if (wasNotFound(ex.getMessage())) return false; + throw ex; + } } @Override public List getInstances() { - var result = messageBroker.request("instance.list", new Document()); - return unwrap(result.mapSuccess(doc -> doc.getList("instances", Document.class) - .stream().map(ServerInstance::fromDocument).toList()), - "failed to get instances"); + var document = messageBroker.request("instance.list", new Document()); + return document.getList("instances", Document.class) + .stream() + .map(ServerInstance::fromDocument) + .toList(); } @Override public @Nullable ServerInstance getInstance(String handle) { - var result = messageBroker.request("instance.get", new Document("handle", handle)); - return unwrap(result.mapSuccess(ServerInstance::fromDocument) - .recover(this::wasNotFound, __ -> null), "failed to get instance " + handle); + var document = messageBroker.request("instance.get", new Document("handle", handle)); + return ServerInstance.fromDocument(document); } private boolean wasNotFound(String failure) { diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/InMemoyMessageBroker.java b/core-lib/src/main/java/de/kentoj/scrowlib/messaging/InMemoyMessageBroker.java new file mode 100644 index 0000000..49716a3 --- /dev/null +++ b/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>> subscribers = new HashMap<>(); + private final Map 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 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); + } + } +} diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/MessageBroker.java b/core-lib/src/main/java/de/kentoj/scrowlib/messaging/MessageBroker.java index 8d33bdd..c5327b8 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/MessageBroker.java +++ b/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 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(); - } +public interface MessageBroker { /** * Runs blocking. * publishes to the message broker + * * @param subject subject to publish to */ - public void publish(String subject, Document document) { - nats.publish(subject, DocumentRepository.toBytes(document)); - } - + void publish(String subject, Document document); + /** - * Runs blocking. - * publishes to the message broker + * subscribes to a subject, consumer may run asynchronously + * * @param subject subject to publish to */ - public void publish(String subject, Result result) { - this.publish(subject, ResultRepository.toDocument(result)); - } + void subscribe(String subject, Consumer consumer); /** - * subscribes to a subject, handler may run asynchronously + * 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 subject to publish to */ - public void subscribe(String subject, Consumer 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)) - ); - }); - } + void handle(String subject, RequestHandler handler); /** * Runs blocking. - * Executes a RPC call: The document is published and this method starts listening for incoming + * Executes an 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 - */ - public Result request(String subject, Document document) { - var response = nats.request(subject, DocumentRepository.toBytes(document)).join(); - return ResultRepository.fromDocument(DocumentRepository.fromMessage(response)); - } - - public static S unwrap(Result result, String message) { - Optional err = result.getFailure(); - if (err.isPresent()) - throw new RuntimeException(message + ": " + err.orElseThrow()); - return result.getSuccess().orElseThrow(); - } + * + * @param subject subject to publish to + * @return RemoteException if the remote throws an exception while handling the request + */ + Document request(String subject, Document document); } diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/NatsMessageBroker.java b/core-lib/src/main/java/de/kentoj/scrowlib/messaging/NatsMessageBroker.java new file mode 100644 index 0000000..7f7f7b3 --- /dev/null +++ b/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 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 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 unwrap(Result result) { + Optional err = result.getFailure(); + if (err.isPresent()) + throw new RemoteException(err.get()); + return result.getSuccess().get(); + } +} diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/RequestHandler.java b/core-lib/src/main/java/de/kentoj/scrowlib/messaging/RequestHandler.java index 2a2b896..be04440 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/RequestHandler.java +++ b/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> handle(Document doc); + Future handle(Document doc); } diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/ResultRepository.java b/core-lib/src/main/java/de/kentoj/scrowlib/messaging/ResultRepository.java index 59a5d74..eb39506 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/ResultRepository.java +++ b/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 result) { + static Document toDocument(Result result) { return result.hasSuccess() ? new Document("data", result.getSuccess().orElseThrow()) : new Document("error", result.getFailure().orElseThrow()); } - public static Result fromDocument(Document doc) { + static Result fromDocument(Document doc) { var error = doc.getString("error"); return error != null ? Results.failure(error) diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/SyncRequestHandler.java b/core-lib/src/main/java/de/kentoj/scrowlib/messaging/SyncRequestHandler.java index bece2d0..ef604f7 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/SyncRequestHandler.java +++ b/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{ - Result handleSync(Document doc); +public interface SyncRequestHandler extends RequestHandler { + Document handleSync(Document doc); @Override - default CompletionStage> handle(Document doc) { + default Future handle(Document doc) { return CompletableFuture.completedFuture(handleSync(doc)); } } diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/player/NetworkPlayer.java b/core-lib/src/main/java/de/kentoj/scrowlib/player/NetworkPlayer.java index eff3c46..7e0b3ff 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/player/NetworkPlayer.java +++ b/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. - * + *

* 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) - * - * 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 {@link MessageBroker} + * @see NetworkPlayerImpl + * @see MessageBroker */ public interface NetworkPlayer extends Audience { diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/DocumentRepository.java b/core-lib/src/main/java/de/kentoj/scrowlib/utils/DocumentRepository.java similarity index 66% rename from core-lib/src/main/java/de/kentoj/scrowlib/messaging/DocumentRepository.java rename to core-lib/src/main/java/de/kentoj/scrowlib/utils/DocumentRepository.java index 213458d..923154f 100644 --- a/core-lib/src/main/java/de/kentoj/scrowlib/messaging/DocumentRepository.java +++ b/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) { - return Document.parse(new String(message.getData())); + public static Document fromBytes(byte[] bytes) { + return Document.parse(new String(bytes)); } @ApiStatus.Obsolete diff --git a/core-lib/src/main/java/de/kentoj/scrowlib/utils/ResultUtils.java b/core-lib/src/main/java/de/kentoj/scrowlib/utils/ResultUtils.java deleted file mode 100644 index 6f36aa0..0000000 --- a/core-lib/src/main/java/de/kentoj/scrowlib/utils/ResultUtils.java +++ /dev/null @@ -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 { -} diff --git a/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/CoreVelocityPlugin.java b/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/CoreVelocityPlugin.java index 6b87f2e..d3e4f33 100644 --- a/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/CoreVelocityPlugin.java +++ b/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(); { diff --git a/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/ScrowAPISurface.java b/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/ScrowAPISurface.java index f8c4a2a..293d029 100644 --- a/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/ScrowAPISurface.java +++ b/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) { - try { - String natsHost = EnvUtils.envOrThrow("HOST_NATS"); - var options = Options.builder() - .server(natsHost) - .pedantic() - .build(); - ScrowAPI.setMessageBroker(new MessageBroker(Nats.connect(options))); - } catch (IOException | InterruptedException e) { - throw new RuntimeException(e); + public static void initScrowAPI(ProxyServer server) { + + var natsHost = System.getenv("HOST_NATS"); + if (natsHost != null) { + try { + var options = Options.builder() + .server(natsHost) + .pedantic() + .build(); + 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(); - props.setProperty("user", "postgres"); - ScrowAPI.setJdbi(Jdbi.create(EnvUtils.envOrThrow("HOST_POSTGRES"), props)); + } else { + log.error("HOST_POSTGRES not set -> Using in-memory database"); } ScrowAPI.setInstanceManager(new InstanceManagerImpl(ScrowAPI.getMessageBroker())); diff --git a/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/network/SendPlayerWatchdog.java b/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/network/SendPlayerWatchdog.java index 0c375ea..213adfb 100644 --- a/core-velocity-impl/src/main/java/de/kentoj/scrow/corevelocity/network/SendPlayerWatchdog.java +++ b/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.CompletionStage; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +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> sendPlayer(UUID playerId, String handle) { - return CompletableFuture.supplyAsync(() -> { + private Future sendPlayer(UUID playerId, String handle) { + 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()); - } catch (InterruptedException | ExecutionException e) { - log.error("failed to send player to {}", handle, e); - return Results.failure(e.getMessage()); - } + var res = player.createConnectionRequest(targetServer).connect().get(); + if (!res.isSuccessful()) + throw new RuntimeException(res.getStatus().name()); + return new Document(); }); } } diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 149e697..e893ca8 100644 --- a/gradle/libs.versions.toml +++ b/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" }