diff --git a/src/main/java/dev/selimaj/session/Session.java b/src/main/java/dev/selimaj/session/Session.java index e0308fe..83c9ee9 100644 --- a/src/main/java/dev/selimaj/session/Session.java +++ b/src/main/java/dev/selimaj/session/Session.java @@ -7,6 +7,7 @@ import dev.selimaj.session.types.SessionResult; import dev.selimaj.session.types.Message; import dev.selimaj.session.types.Method; import dev.selimaj.session.types.MethodHandler; +import dev.selimaj.session.types.NotificationHandler; import java.net.URI; import java.net.http.HttpClient; @@ -69,10 +70,11 @@ public final class Session { } public void onRequest(Method method, - MethodHandler handler) { - MethodHandler wrapper = (id, value) -> { + MethodHandler handler) { + MethodHandler wrapper = (id, value) -> { try { - return handler.handle(id, listener.mapper.treeToValue(value, method.getReqClass())); + return handler.handle(id, listener.mapper.treeToValue(value, method.getReqClass())) + .intoJSON(listener.mapper); } catch (Exception e) { return SessionResult.error(TextNode.valueOf(e.getMessage())); } @@ -81,6 +83,22 @@ public final class Session { this.listener.methods.put(method.getName(), wrapper); } + public void onNotification(Method method, + NotificationHandler handler) { + NotificationHandler wrapper = (id, value) -> { + return CompletableFuture.supplyAsync(() -> { + try { + return handler.handle(id, listener.mapper.treeToValue(value, method.getReqClass())).get() + .intoJSON(listener.mapper); + } catch (Exception e) { + return SessionResult.error(TextNode.valueOf(e.getMessage())); + } + }); + }; + + this.listener.notificationHandlers.put(method.getName(), wrapper); + } + public void close() throws Exception { ws.sendClose(0, "Session closed"); } diff --git a/src/main/java/dev/selimaj/session/SessionListener.java b/src/main/java/dev/selimaj/session/SessionListener.java index fc7546b..b7efdbe 100644 --- a/src/main/java/dev/selimaj/session/SessionListener.java +++ b/src/main/java/dev/selimaj/session/SessionListener.java @@ -3,6 +3,7 @@ package dev.selimaj.session; import java.net.http.WebSocket; import java.nio.ByteBuffer; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; @@ -12,9 +13,11 @@ import com.fasterxml.jackson.databind.ObjectMapper; import dev.selimaj.session.types.Message; import dev.selimaj.session.types.MethodHandler; +import dev.selimaj.session.types.NotificationHandler; public class SessionListener implements WebSocket.Listener { - final ConcurrentHashMap> methods = new ConcurrentHashMap<>(); + final ConcurrentHashMap> methods = new ConcurrentHashMap<>(); + final ConcurrentHashMap> notificationHandlers = new ConcurrentHashMap<>(); final ObjectMapper mapper = new ObjectMapper(); private final ConcurrentHashMap> pending = new ConcurrentHashMap<>(); @@ -32,20 +35,19 @@ public class SessionListener implements WebSocket.Listener { } if (msg instanceof Message.Request r) { - MethodHandler handler = methods.get(r.method()); + MethodHandler handler = methods.get(r.method()); if (handler != null) { - handler.handle(r.id(), r.data()) - .thenAccept(res -> { - try { - if (res.isError()) { - respondError(ws, r.id(), res.value()); - } else { - respond(ws, r.id(), res.value()); - } - } catch (Exception ignored) { - } - }); + try { + var res = handler.handle(r.id(), r.data()); + + if (res.isError()) { + respondError(ws, r.id(), res.getJsonNode(mapper)); + } else { + respond(ws, r.id(), res.getJsonNode(mapper)); + } + } catch (Exception ignored) { + } } } else if (msg instanceof Message.Response r) { CompletableFuture fut = pending.remove(r.id()); @@ -101,28 +103,36 @@ public class SessionListener implements WebSocket.Listener { send(ws, new Message.Notification(method, mapper.valueToTree(data))); } - CompletableFuture request( + public CompletableFuture request( WebSocket ws, String method, Req data, - Class resType) throws Exception { + Class resType) { int id = this.id.incrementAndGet(); - CompletableFuture fut = new CompletableFuture<>(); - pending.put(id, fut); - - send(ws, new Message.Request( - id, - method, - mapper.valueToTree(data))); - - return fut.thenApply(json -> { - try { - return mapper.treeToValue(json, resType); - } catch (Exception e) { - throw new RuntimeException(e); - } - }); + return CompletableFuture + .runAsync(() -> { + try { + send(ws, new Message.Request( + id, + method, + mapper.valueToTree(data))); + } catch (Exception e) { + throw new CompletionException(e); + } + }) + .thenCompose(v -> { + CompletableFuture fut = new CompletableFuture<>(); + pending.put(id, fut); + return fut; + }) + .thenApply(json -> { + try { + return mapper.treeToValue(json, resType); + } catch (Exception e) { + throw new CompletionException(e); + } + }); } } diff --git a/src/main/java/dev/selimaj/session/types/MethodHandler.java b/src/main/java/dev/selimaj/session/types/MethodHandler.java index 9ac8315..21cbc90 100644 --- a/src/main/java/dev/selimaj/session/types/MethodHandler.java +++ b/src/main/java/dev/selimaj/session/types/MethodHandler.java @@ -1,8 +1,6 @@ package dev.selimaj.session.types; -import java.util.concurrent.CompletableFuture; - @FunctionalInterface -public interface MethodHandler { - CompletableFuture handle(int id, Req data); +public interface MethodHandler { + SessionResult handle(int id, Req data); } diff --git a/src/main/java/dev/selimaj/session/types/NotificationHandler.java b/src/main/java/dev/selimaj/session/types/NotificationHandler.java new file mode 100644 index 0000000..a7dced9 --- /dev/null +++ b/src/main/java/dev/selimaj/session/types/NotificationHandler.java @@ -0,0 +1,8 @@ +package dev.selimaj.session.types; + +import java.util.concurrent.CompletableFuture; + +@FunctionalInterface +public interface NotificationHandler { + CompletableFuture> handle(int id, Req data); +} diff --git a/src/main/java/dev/selimaj/session/types/SessionResult.java b/src/main/java/dev/selimaj/session/types/SessionResult.java index a32687a..93f090c 100644 --- a/src/main/java/dev/selimaj/session/types/SessionResult.java +++ b/src/main/java/dev/selimaj/session/types/SessionResult.java @@ -1,15 +1,22 @@ package dev.selimaj.session.types; -import java.util.concurrent.CompletableFuture; - import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; -public record SessionResult(boolean isError, JsonNode value) { - public static CompletableFuture ok(JsonNode value) { - return CompletableFuture.completedFuture(new SessionResult(false, value)); +public record SessionResult(boolean isError, Object value) { + public static SessionResult ok(T value) { + return new SessionResult<>(false, value); } - public static CompletableFuture error(JsonNode value) { - return CompletableFuture.completedFuture(new SessionResult(false, value)); + public static SessionResult error(E err) { + return new SessionResult<>(false, err); + } + + public SessionResult intoJSON(ObjectMapper mapper) { + return new SessionResult<>(this.isError, mapper.valueToTree(this.value)); + } + + public JsonNode getJsonNode(ObjectMapper mapper) { + return mapper.valueToTree(this.value); } }