Add http endpoint events with SSE

diff --git a/src/main/java/org/asamk/signal/http/HttpServerHandler.java b/src/main/java/org/asamk/signal/http/HttpServerHandler.java
index 32000a1f4db86096ea155ea86432b601ab743b8a..f7f2768d6bffdf21312d7557af758982ecce2a6e 100644 (file)
--- a/src/main/java/org/asamk/signal/http/HttpServerHandler.java
+++ b/src/main/java/org/asamk/signal/http/HttpServerHandler.java
@@ -5,19 +5,25 @@ import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpServer;
 
 import org.asamk.signal.commands.Commands;
+import org.asamk.signal.json.JsonReceiveMessageHandler;
 import org.asamk.signal.jsonrpc.JsonRpcReader;
 import org.asamk.signal.jsonrpc.JsonRpcResponse;
 import org.asamk.signal.jsonrpc.JsonRpcSender;
 import org.asamk.signal.jsonrpc.SignalJsonRpcCommandHandler;
 import org.asamk.signal.manager.Manager;
 import org.asamk.signal.manager.MultiAccountManager;
+import org.asamk.signal.manager.api.Pair;
+import org.asamk.signal.manager.util.Utils;
 import org.asamk.signal.util.Util;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class HttpServerHandler {
 
@@ -28,15 +34,21 @@ public class HttpServerHandler {
     private final InetSocketAddress address;
 
     private final SignalJsonRpcCommandHandler commandHandler;
+    private final MultiAccountManager c;
+    private final Manager m;
 
     public HttpServerHandler(final InetSocketAddress address, final Manager m) {
         this.address = address;
         commandHandler = new SignalJsonRpcCommandHandler(m, Commands::getCommand);
+        this.c = null;
+        this.m = m;
     }
 
     public HttpServerHandler(final InetSocketAddress address, final MultiAccountManager c) {
         this.address = address;
         commandHandler = new SignalJsonRpcCommandHandler(c, Commands::getCommand);
+        this.c = c;
+        this.m = null;
     }
 
     public void init() throws IOException {
@@ -46,6 +58,7 @@ public class HttpServerHandler {
         server.setExecutor(Executors.newFixedThreadPool(10));
 
         server.createContext("/api/v1/rpc", this::handleRpcEndpoint);
+        server.createContext("/api/v1/events", this::handleEventsEndpoint);
 
         server.start();
     }
@@ -110,4 +123,112 @@ public class HttpServerHandler {
                     httpExchange);
         }
     }
+
+    private void handleEventsEndpoint(HttpExchange httpExchange) throws IOException {
+        if (!"/api/v1/events".equals(httpExchange.getRequestURI().getPath())) {
+            sendResponse(404, null, httpExchange);
+            return;
+        }
+        if (!"GET".equals(httpExchange.getRequestMethod())) {
+            sendResponse(405, null, httpExchange);
+            return;
+        }
+
+        try {
+            final var queryString = httpExchange.getRequestURI().getQuery();
+            final var query = queryString == null ? Map.<String, String>of() : Utils.getQueryMap(queryString);
+
+            List<Manager> managers = getManagerFromQuery(query);
+            if (managers == null) {
+                sendResponse(400, null, httpExchange);
+                return;
+            }
+
+            httpExchange.getResponseHeaders().add("Content-Type", "text/event-stream");
+            httpExchange.sendResponseHeaders(200, 0);
+            final var sender = new ServerSentEventSender(httpExchange.getResponseBody());
+
+            final var shouldStop = new AtomicBoolean(false);
+            final var handlers = subscribeReceiveHandlers(managers, sender, () -> {
+                shouldStop.set(true);
+                synchronized (this) {
+                    this.notify();
+                }
+            });
+
+            try {
+                while (true) {
+                    synchronized (this) {
+                        wait(15_000);
+                    }
+                    if (shouldStop.get()) {
+                        break;
+                    }
+
+                    try {
+                        sender.sendKeepAlive();
+                    } catch (IOException e) {
+                        break;
+                    }
+                }
+            } finally {
+                for (final var pair : handlers) {
+                    unsubscribeReceiveHandler(pair);
+                }
+                try {
+                    httpExchange.getResponseBody().close();
+                } catch (IOException ignored) {
+                }
+            }
+        } catch (Throwable aEx) {
+            logger.error("Failed to process request.", aEx);
+            sendResponse(500, null, httpExchange);
+        }
+    }
+
+    private List<Manager> getManagerFromQuery(final Map<String, String> query) {
+        List<Manager> managers;
+        if (m != null) {
+            managers = List.of(m);
+        } else {
+            final var account = query.get("account");
+            if (account == null || account.isEmpty()) {
+                managers = c.getManagers();
+            } else {
+                final var manager = c.getManager(account);
+                if (manager == null) {
+                    return null;
+                }
+                managers = List.of(manager);
+            }
+        }
+        return managers;
+    }
+
+    private List<Pair<Manager, Manager.ReceiveMessageHandler>> subscribeReceiveHandlers(
+            final List<Manager> managers, final ServerSentEventSender sender, Callable unsubscribe
+    ) {
+        return managers.stream().map(m1 -> {
+            final var receiveMessageHandler = new JsonReceiveMessageHandler(m1, s -> {
+                try {
+                    sender.sendEvent(null, "receive", List.of(objectMapper.writeValueAsString(s)));
+                } catch (IOException e) {
+                    unsubscribe.call();
+                }
+            });
+            m1.addReceiveHandler(receiveMessageHandler);
+            return new Pair<>(m1, (Manager.ReceiveMessageHandler) receiveMessageHandler);
+        }).toList();
+    }
+
+    private void unsubscribeReceiveHandler(final Pair<Manager, Manager.ReceiveMessageHandler> pair) {
+        final var m = pair.first();
+        final var handler = pair.second();
+        m.removeReceiveHandler(handler);
+    }
+
+    private interface Callable {
+
+        void call();
+    }
 }

diff --git a/src/main/java/org/asamk/signal/http/ServerSentEventSender.java b/src/main/java/org/asamk/signal/http/ServerSentEventSender.java
new file mode 100644 (file)
index 0000000..b33d331
--- /dev/null
+++ b/src/main/java/org/asamk/signal/http/ServerSentEventSender.java
@@ -0,0 +1,55 @@
+package org.asamk.signal.http;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/**
+ * This class send Server-sent events payload to an OutputStream.
+ * See <a href="https://html.spec.whatwg.org/multipage/server-sent-events.html">spec</a>
+ */
+public class ServerSentEventSender {
+
+    private final BufferedWriter writer;
+
+    public ServerSentEventSender(final OutputStream outputStream) {
+        this.writer = new BufferedWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8));
+    }
+
+    /**
+     * @param id    Event id
+     * @param event Event type
+     * @param data  Event data, each entry must not contain newline chars.
+     */
+    public synchronized void sendEvent(String id, String event, List<String> data) throws IOException {
+        if (id != null) {
+            writer.write("id:");
+            writer.write(id);
+            writer.write("\n");
+        }
+        if (event != null) {
+            writer.write("event:");
+            writer.write(event);
+            writer.write("\n");
+        }
+        if (data.size() == 0) {
+            writer.write("data\n");
+        } else {
+            for (final var d : data) {
+                writer.write("data:");
+                writer.write(d);
+                writer.write("\n");
+            }
+        }
+        writer.write("\n");
+        writer.flush();
+    }
+
+    public synchronized void sendKeepAlive() throws IOException {
+        writer.write(":\n");
+        writer.flush();
+    }
+}