Backend: Server-Sent Events for handlers

docs/demos/backend/src/main/java/com/codenameone/developerguide/backend/Streaming.java

@@ -24,6 +24,7 @@
 
 import com.codename1.backend.HttpProxy;
 import com.codename1.backend.HttpServer;
+import com.codename1.backend.Sse;
 import com.codename1.backend.Web;
 import com.codename1.backend.annotations.GetMapping;
 import com.codename1.backend.annotations.PostMapping;
@@ -77,6 +78,50 @@ public void writeTo(OutputStream out) throws IOException {
     }
     // end::backend-web-download-stream[]
 
+    // tag::backend-web-sse[]
+    @GetMapping("/prices")
+    public HttpServer.Response prices(HttpServer.Request request) {
+        return Sse.stream(request, events -> {
+            long last = parse(events.lastEventId());  // where a reconnect left off
+            for (int idle = 0; idle < 600; idle++) {
+                long newest = PriceBoard.newest();
+                if (newest > last) {
+                    events.event("price").id(Long.toString(newest)).json(PriceBoard.since(last));
+                    last = newest;
+                } else if (idle % 15 == 0) {
+                    events.ping();                    // keep proxies from closing it
+                }
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException stopped) {
+                    return;
+                }
+            }
+        });
+    }
+    // end::backend-web-sse[]
+
+    private static long parse(String id) {
+        // The header is whatever the client sent; a bad one starts over
+        // rather than failing a stream whose headers are already out.
+        try {
+            return id == null ? 0 : Math.max(0, Long.parseLong(id.trim()));
+        } catch (NumberFormatException notOurs) {
+            return 0;
+        }
+    }
+
+    /// Stands in for wherever the application keeps its prices.
+    static final class PriceBoard {
+        static long newest() {
+            return 0;
+        }
+
+        static Map<String, Object> since(long version) {
+            return Collections.emptyMap();
+        }
+    }
+
     // tag::backend-web-exchange[]
     static long upload(InputStream file, long length) throws IOException {
         Web.Exchange exchange = Web.open("PUT", "https://storage.internal/blobs/1",

docs/developer-guide/Backend-Web.asciidoc

@@ -437,6 +437,34 @@ something open until it runs, such as a connection or a file, should also
 implement `java.io.Closeable`. In those cases the server calls `close()`
 instead.
 
+==== Server-Sent Events
+
+For a feed the server pushes and the client only reads -- tokens relayed from a
+language model as they arrive, prices, progress -- an event stream is
+simpler than a WebSocket: it's one ordinary `GET`, the browser's `EventSource`
+reconnects by itself, and it passes through any proxy that passes HTTP.
+`Sse.stream` answers with `text/event-stream` and hands the handler an
+`Sse.Events`. Each `data` or `json` call sends one event and flushes it, so the
+client has it before the next one is produced.
+
+[source,java,role=cn1-no-try]
+----
+include::../demos/backend/src/main/java/com/codenameone/developerguide/backend/Streaming.java[tag=backend-web-sse,indent=0]
+----
+
+`event`, `id` and `retry` set fields of the next event. A client that
+reconnects sends the last id it saw, which `lastEventId()` returns, so the
+stream can resume after it. The producer runs on the worker that served the
+request and the stream ends when it returns. A client that goes away makes the
+next send throw. Proxies and load balancers close a connection that has been
+idle for about a minute, so a producer that waits for news calls `ping()`
+from time to time; the client never sees it. Wait with `Thread.sleep`, not
+`Object.wait`.
+
+To relay an event stream from another server unchanged, `HttpProxy` already
+streams it event by event, as it does any upstream body of undeclared
+length. Compression never applies to a streamed response.
+
 ==== Compression
 
 `cn1.server.compression.enabled=true` turns on gzip for responses, with the two

maven/backend/src/test/java/com/codename1/backend/SseTest.java

@@ -0,0 +1,174 @@
+/*
+ * Copyright (c) 2012, Codename One and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Codename One designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Codename One through http://www.codenameone.com/ if you
+ * need additional information or have any questions.
+ */
+package com.codename1.backend;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/// Server-Sent Events: the framing, and real sockets showing each event reaches
+/// the client before the producer has finished.
+class SseTest {
+    private Backend backend;
+
+    @AfterEach
+    void stop() {
+        if (backend != null) {
+            backend.stop();
+        }
+    }
+
+    private static String frame(Sse.Producer producer) throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        producer.produce(new Sse.Events(out, null));
+        return new String(out.toByteArray(), StandardCharsets.UTF_8);
+    }
+
+    @Test
+    void framesFieldsAndMultiLineData() throws IOException {
+        assertEquals("data: hi\n\n", frame(e -> e.data("hi")));
+        assertEquals("event: tick\nid: 7\nretry: 1500\ndata: a\ndata: b\ndata: c\ndata:\ndata: d\n\n",
+                frame(e -> e.event("tick").id("7").retry(1500).data("a\nb\r\nc\r\rd")));
+        assertEquals("data:\n\n", frame(e -> e.data("")));
+        // Fields belong to one event only.
+        assertEquals("event: x\ndata: 1\n\ndata: 2\n\n", frame(e -> {
+            e.event("x").data("1");
+            e.data("2");
+        }));
+        assertEquals(": note\n\n:\n\n", frame(e -> {
+            e.comment("note");
+            e.ping();
+        }));
+        assertEquals("id: 9\n\n", frame(e -> e.id("9").send()));
+        assertEquals("data: caf\u00e9 \u2603\n\n", frame(e -> e.data("caf\u00e9 \u2603")));
+        Map value = new HashMap();
+        value.put("text", "line\nbreak");
+        assertEquals("data: {\"text\":\"line\\nbreak\"}\n\n", frame(e -> e.json(value)));
+    }
+
+    @Test
+    void refusesFieldsThatWouldForgeAnother() {
+        Sse.Events events = new Sse.Events(new ByteArrayOutputStream(), null);
+        assertThrows(IllegalArgumentException.class, () -> events.event("a\ndata: forged"));
+        assertThrows(IllegalArgumentException.class, () -> events.id("1\r2"));
+        assertThrows(IllegalArgumentException.class, () -> events.id("1\u00002"));
+        assertThrows(IllegalArgumentException.class, () -> events.retry(-1));
+    }
+
+    /// Fresh per test, so a rerun still proves the first event arrived early.
+    private final CountDownLatch firstRead = new CountDownLatch(1);
+    private HttpProxy proxy;
+
+    private int start(boolean compression) throws Exception {
+        ServerSocket s = new ServerSocket(0);
+        int port = s.getLocalPort();
+        s.close();
+        Properties settings = new Properties();
+        settings.setProperty(Config.SERVER_PORT, String.valueOf(port));
+        if (compression) {
+            settings.setProperty(Compression.ENABLED, "true");
+            settings.setProperty(Compression.MIN_SIZE, "0");
+            settings.setProperty(Compression.MIME_TYPES, "text/event-stream");
+        }
+        proxy = new HttpProxy("http://127.0.0.1:" + port);
+        backend = Backend.builder(Config.of(settings, "test")).quiet().handler(request ->
+                request.getTarget().startsWith("/proxy")
+                ? proxy.forward(request, "/events")
+                : Sse.stream(request, events -> {
+                    events.id("1").data("first " + events.lastEventId());
+                    // Held until the client has read the first event: a server that
+                    // buffered to the end would deadlock here and the test time out.
+                    try {
+                        if (!firstRead.await(20, TimeUnit.SECONDS)) {
+                            throw new IOException("first event never reached the client");
+                        }
+                    } catch (InterruptedException err) {
+                        throw new IOException(err.toString());
+                    }
+                    events.event("done").data("second");
+                })).start();
+        return port;
+    }
+
+    private String exchange(int port, String request, boolean signal) throws IOException {
+        try (Socket socket = new Socket("127.0.0.1", port)) {
+            socket.setSoTimeout(30000);
+            OutputStream out = socket.getOutputStream();
+            out.write(request.getBytes(StandardCharsets.US_ASCII));
+            out.flush();
+            InputStream in = socket.getInputStream();
+            StringBuilder seen = new StringBuilder();
+            byte[] buffer = new byte[4096];
+            int n;
+            while ((n = in.read(buffer)) > 0) {
+                seen.append(new String(buffer, 0, n, StandardCharsets.UTF_8));
+                if (signal && seen.indexOf("first ") >= 0 && seen.indexOf("\n\n", seen.indexOf("first ")) >= 0) {
+                    firstRead.countDown();
+                }
+            }
+            return seen.toString();
+        }
+    }
+
+    @Test
+    void eachEventIsFlushedAsWritten() throws Exception {
+        int port = start(true);
+        String reply = exchange(port, "GET /events HTTP/1.1\r\nHost: x\r\nAccept-Encoding: gzip\r\n"
+                + "Last-Event-ID: 41\r\nConnection: close\r\n\r\n", true);
+        assertTrue(reply.startsWith("HTTP/1.1 200"), reply);
+        assertTrue(reply.contains("Content-Type: text/event-stream; charset=utf-8"), reply);
+        assertTrue(reply.contains("Cache-Control: no-cache"), reply);
+        assertTrue(reply.contains("Transfer-Encoding: chunked"), reply);
+        // Compression is on for this type with no minimum, and still not applied.
+        assertFalse(reply.toLowerCase().contains("content-encoding"), reply);
+        assertTrue(reply.contains("id: 1\ndata: first 41\n\n"), reply);
+        assertTrue(reply.contains("event: done\ndata: second\n\n"), reply);
+    }
+
+    @Test
+    void theProxyRelaysEachEventAsItArrives() throws Exception {
+        int port = start(false);
+        String reply = exchange(port, "GET /proxy HTTP/1.1\r\nHost: x\r\n"
+                + "Last-Event-ID: 5\r\nConnection: close\r\n\r\n", true);
+        assertTrue(reply.startsWith("HTTP/1.1 200"), reply);
+        assertTrue(reply.contains("id: 1\ndata: first 5\n\n"), reply);
+        assertTrue(reply.contains("event: done\ndata: second\n\n"), reply);
+    }
+}

vm/backend/src/com/codename1/backend/Compression.java

@@ -85,6 +85,10 @@ void apply(HttpServer.Request request, HttpServer.Response response) throws IOEx
         // representation: compressing it would leave the offsets describing bytes
         // the client never receives, and a resumed download assembled corrupt.
         if (response == null || response.fileFd >= 0
+                // A streamed body is written after this runs; marking it gzip would
+                // label identity bytes as compressed. Event streams also need every
+                // write to reach the client at once, which gzip would hold back.
+                || response.bodyWriter != null
                 || response.status == 204 || response.status == 304 || response.status == 206
                 || response.hasHeader("Content-Range")
                 || response.hasHeader("Content-Encoding") || !compressible(response.contentType)) {

vm/backend/src/com/codename1/backend/HttpProxy.java

@@ -326,7 +326,7 @@ private static HttpServer.Response respond(HttpServer.Request request,
         // as a stream so that none is invented: the server never runs a HEAD's
         // writer, and closes it -- ending the exchange -- instead.
         HttpServer.Response response = HttpServer.Response.stream(status, contentType,
-                new Relay(exchange));
+                new Relay(exchange, described < 0 || isEventStream(contentType)));
         response.extraHeaders = out;
         return response;
     }
@@ -336,15 +336,22 @@ private static HttpServer.Response respond(HttpServer.Request request,
     /// HTTP/2 stream it could not open), and the exchange must end either way.
     private static final class Relay implements HttpServer.BodyWriter, Closeable {
         private final Web.Exchange exchange;
+        private final boolean eachRead;
 
-        Relay(Web.Exchange exchange) {
+        /// - `eachRead`: pass every upstream read on at once. True for an event
+        ///   stream and for any body of undeclared length (a token feed, a log
+        ///   tail): those trickle small writes, and holding them until the 16 KB
+        ///   buffer filled delayed every event by minutes. A body of known length
+        ///   is a download, and keeps whole 16 KB chunks.
+        Relay(Web.Exchange exchange, boolean eachRead) {
             this.exchange = exchange;
+            this.eachRead = eachRead;
         }
 
         @Override
         public void writeTo(OutputStream output) throws IOException {
             try {
-                copy(exchange.getResponseBody(), output);
+                copy(exchange.getResponseBody(), output, eachRead);
             } finally {
                 // Complete, or cancelled: either way the upstream is done, and on
                 // a client that went away this is what stops it sending.
@@ -554,11 +561,29 @@ static String resolve(String current, String location) {
     }
 
     private static void copy(InputStream in, OutputStream out) throws IOException {
+        copy(in, out, false);
+    }
+
+    /// Whether a Content-Type names `text/event-stream`, whatever its parameters.
+    static boolean isEventStream(String contentType) {
+        if (contentType == null) {
+            return false;
+        }
+        int semi = contentType.indexOf(';');
+        String type = (semi < 0 ? contentType : contentType.substring(0, semi)).trim();
+        return "text/event-stream".equalsIgnoreCase(type);
+    }
+
+    private static void copy(InputStream in, OutputStream out, boolean flushEachRead)
+            throws IOException {
         byte[] buffer = new byte[16 * 1024];
         int n;
         while ((n = in.read(buffer, 0, buffer.length)) >= 0) {
             if (n > 0) {
                 out.write(buffer, 0, n);
+                if (flushEachRead) {
+                    out.flush();
+                }
             }
         }
     }