Simple session structure

src/main/java/dev/selimaj/session/Session.java

@@ -1,6 +1,9 @@
 package dev.selimaj.session;
 
-import java.util.concurrent.*;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import dev.selimaj.session.types.Message;
+
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.WebSocket;
@@ -8,12 +11,10 @@ import java.net.http.WebSocket;
 public final class Session {
 
     private final WebSocket ws;
-    private final ExecutorService executor;
     private final SessionListener listener;
 
-    public Session(WebSocket ws, SessionListener listener, ExecutorService executor) {
+    public Session(WebSocket ws, SessionListener listener) {
         this.ws = ws;
-        this.executor = executor;
         this.listener = listener;
     }
 
@@ -22,16 +23,29 @@ public final class Session {
 
         WebSocket ws = HttpClient.newHttpClient()
                 .newWebSocketBuilder()
-                .buildAsync(URI.create("ws://example.com/echo"), listener)
+                .buildAsync(URI.create(uri), listener)
                 .join();
 
-        return new Session(ws, listener, Executors.newCachedThreadPool());
+        return new Session(ws, listener);
     }
 
-    public void startReceiver() {
+    public void send(Message msg) throws Exception {
-        executor.execute(() -> {
+        listener.send(ws, msg);
-
-        });
     }
 
+    public void respond(int to, JsonNode val) throws Exception {
+        listener.respond(ws, to, val);
+    }
+
+    public void respondError(int to, JsonNode val) throws Exception {
+        listener.respondError(ws, to, val);
+    }
+
+    public void notify(String method, Object data) throws Exception {
+        listener.notify(ws, method, data);
+    }
+
+    public void close() throws Exception {
+        ws.sendClose(0, "Session closed");
+    }
 }

src/main/java/dev/selimaj/session/SessionListener.java

@@ -1,10 +1,76 @@
 package dev.selimaj.session;
 
 import java.net.http.WebSocket;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import dev.selimaj.session.types.Message;
 import dev.selimaj.session.types.MethodHandler;
 
 public class SessionListener implements WebSocket.Listener {
     public final ConcurrentHashMap<String, MethodHandler> methods = new ConcurrentHashMap<>();
+
+    private final ObjectMapper mapper = new ObjectMapper();
+    private final ConcurrentHashMap<Integer, CompletableFuture<JsonNode>> pending = new ConcurrentHashMap<>();
+
+    @Override
+    public CompletionStage<?> onText(WebSocket ws, CharSequence data, boolean last) {
+        Message msg = null;
+
+        try {
+            msg = mapper.readValue(data.toString(), Message.class);
+        } catch (Exception e) {
+            e.printStackTrace();
+            return null;
+        }
+
+        if (msg instanceof Message.Request r) {
+            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) {
+                            }
+                        });
+            }
+        } else if (msg instanceof Message.Response r) {
+            CompletableFuture<JsonNode> fut = pending.remove(r.id());
+            if (fut != null)
+                fut.complete(r.result());
+        } else if (msg instanceof Message.ErrorResponse r) {
+            CompletableFuture<JsonNode> fut = pending.remove(r.id());
+            if (fut != null)
+                fut.completeExceptionally(
+                        new RuntimeException(r.error().toString()));
+        }
+
+        return null;
+    }
+
+    void send(WebSocket ws, Message msg) throws Exception {
+        ws.sendText(mapper.writeValueAsString(msg), true);
+    }
+
+    void respond(WebSocket ws, int to, JsonNode val) throws Exception {
+        send(ws, new Message.Response(to, val));
+    }
+
+    void respondError(WebSocket ws, int to, JsonNode val) throws Exception {
+        send(ws, new Message.ErrorResponse(to, val));
+    }
+
+    void notify(WebSocket ws, String method, Object data) throws Exception {
+        send(ws, new Message.Notification(method, mapper.valueToTree(data)));
+    }
 }