diff --git a/docs/demos/backend/src/main/java/com/codenameone/developerguide/backend/Streaming.java b/docs/demos/backend/src/main/java/com/codenameone/developerguide/backend/Streaming.java index e507cb64a06..09719045121 100644 --- a/docs/demos/backend/src/main/java/com/codenameone/developerguide/backend/Streaming.java +++ b/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 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", diff --git a/docs/developer-guide/Backend-Web.asciidoc b/docs/developer-guide/Backend-Web.asciidoc index 14b9841800b..e3f78d6a31e 100644 --- a/docs/developer-guide/Backend-Web.asciidoc +++ b/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 diff --git a/maven/backend/src/test/java/com/codename1/backend/SseTest.java b/maven/backend/src/test/java/com/codename1/backend/SseTest.java new file mode 100644 index 00000000000..05291fcbb01 --- /dev/null +++ b/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); + } +} diff --git a/vm/backend/src/com/codename1/backend/Compression.java b/vm/backend/src/com/codename1/backend/Compression.java index 74bbbd45b13..b2bedea2f54 100644 --- a/vm/backend/src/com/codename1/backend/Compression.java +++ b/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)) { diff --git a/vm/backend/src/com/codename1/backend/HttpProxy.java b/vm/backend/src/com/codename1/backend/HttpProxy.java index c75a963c5e7..68b4b825585 100644 --- a/vm/backend/src/com/codename1/backend/HttpProxy.java +++ b/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(); + } } } } diff --git a/vm/backend/src/com/codename1/backend/Sse.java b/vm/backend/src/com/codename1/backend/Sse.java new file mode 100644 index 00000000000..ef657109587 --- /dev/null +++ b/vm/backend/src/com/codename1/backend/Sse.java @@ -0,0 +1,263 @@ +/* + * 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 java.io.IOException; +import java.io.OutputStream; + +/// Server-Sent Events (`text/event-stream`) from a handler: a response that stays +/// open and delivers each event to the client the moment it is written. +/// +/// ```java +/// return Sse.stream(request, events -> { +/// for (String token : model.tokens(prompt)) { +/// events.data(token); // on the wire now, not at the end +/// } +/// events.event("done").data("{}"); +/// }); +/// ``` +/// +/// The producer runs on the worker that served the request, after the headers +/// are sent, and the stream ends when it returns. It blocks on its own source -- +/// an upstream stream, a queue it polls -- and should call [Events#ping()] when +/// it has had nothing to send for a while: proxies and load balancers close a +/// connection that is idle for about a minute. Wait with `Thread.sleep`, never +/// `Object.wait`, which blocks the whole host under virtual threads. +/// +/// A client that goes away makes the next write throw `IOException`; let it +/// propagate so the producer's try-with-resources cleanup runs. A producer that +/// opens something before it runs should also implement `java.io.Closeable`: the +/// server calls `close()` instead of the producer when no body is sent (HEAD). +/// +/// To relay an upstream event stream unchanged, [HttpProxy] already streams it. +/// +/// This is the blocking-producer model, not Spring's `SseEmitter`: nothing here +/// hands events from another thread to the connection. +public final class Sse { + /// The media type of an event stream. + public static final String CONTENT_TYPE = "text/event-stream; charset=utf-8"; + + private Sse() { + } + + /// Writes the events of one stream. + public interface Producer { + void produce(Events events) throws IOException; + } + + /// An event stream answered with status 200. + public static HttpServer.Response stream(HttpServer.Request request, Producer producer) { + if (producer == null) { + throw new NullPointerException("producer"); + } + // Read now: a body writer must not keep the Request. + String lastEventId = request == null ? null : request.getHeader("Last-Event-ID"); + return HttpServer.Response.stream(200, CONTENT_TYPE, new Writer(producer, lastEventId)) + .header("Cache-Control", "no-cache") + // nginx buffers a proxied response by default, which holds events back. + .header("X-Accel-Buffering", "no"); + } + + private static final class Writer implements HttpServer.BodyWriter, java.io.Closeable { + private final Producer producer; + private final String lastEventId; + + Writer(Producer producer, String lastEventId) { + this.producer = producer; + this.lastEventId = lastEventId; + } + + @Override + public void writeTo(OutputStream output) throws IOException { + producer.produce(new Events(output, lastEventId)); + } + + @Override + public void close() throws IOException { + if (producer instanceof java.io.Closeable) { + ((java.io.Closeable) producer).close(); + } + } + } + + /// The open stream. `event`, `id` and `retry` set fields of the NEXT event, + /// which [#data(String)], [#json(Object)] or [#send()] then writes and flushes. + public static final class Events { + private final OutputStream out; + private final String lastEventId; + private final ByteSink sink = new ByteSink(256); + private final ByteSink json = new ByteSink(256); + private String event; + private String id; + private long retry = -1; + + Events(OutputStream out, String lastEventId) { + this.out = out; + this.lastEventId = lastEventId; + } + + /// The `Last-Event-ID` the client reconnected with, or null on a first + /// connection. Resume after it. + public String lastEventId() { + return lastEventId; + } + + /// Names the next event; a client listens for it with + /// `addEventListener(name, ...)`. Unnamed events are `message`. + public Events event(String name) { + event = checkField("event", name); + return this; + } + + /// The id of the next event, which the client sends back as + /// `Last-Event-ID` when it reconnects. + public Events id(String eventId) { + String checked = checkField("id", eventId); + if (checked != null && checked.indexOf('\0') >= 0) { + // A browser ignores an id containing NUL, so it would never come back. + throw new IllegalArgumentException("an event id cannot contain NUL"); + } + id = checked; + return this; + } + + /// How long the client waits before reconnecting, sent with the next event. + public Events retry(long millis) { + if (millis < 0) { + throw new IllegalArgumentException("retry " + millis + " is negative"); + } + retry = millis; + return this; + } + + /// Sends one event with this data. Each line of a multi-line value + /// becomes its own `data:` field, which the client joins back with `\n`. + public void data(String data) throws IOException { + sink.reset(); + putFields(); + putData(data == null ? "" : data); + finish(); + } + + /// Sends one event whose data is `value` as JSON, encoded straight into + /// the buffer. JSON has no raw line breaks, so it is always one line. + public void json(Object value) throws IOException { + sink.reset(); + putFields(); + json.reset(); + Json.write(value, json); + sink.putAscii("data: "); + sink.put(json.bytes(), 0, json.length()); + sink.put('\n'); + finish(); + } + + /// Sends the pending `event`/`id`/`retry` fields with no data. A browser + /// dispatches nothing for it, but records the id and the retry. + public void send() throws IOException { + sink.reset(); + putFields(); + finish(); + } + + /// A comment line, which every client ignores. + public void comment(String text) throws IOException { + sink.reset(); + putLines(":", text == null ? "" : text); + sink.put('\n'); + write(); + } + + /// Keeps an idle stream open through proxies; the client sees nothing. + public void ping() throws IOException { + sink.reset(); + sink.putAscii(":\n\n"); + write(); + } + + private void putFields() { + if (event != null) { + sink.putAscii("event: "); + sink.putUtf8(event); + sink.put('\n'); + } + if (id != null) { + sink.putAscii("id: "); + sink.putUtf8(id); + sink.put('\n'); + } + if (retry >= 0) { + sink.putAscii("retry: "); + sink.putNumber(retry); + sink.put('\n'); + } + event = null; + id = null; + retry = -1; + } + + private void putData(String data) { + putLines("data:", data); + } + + /// One `prefix value` line per line of `text`, split at CR, LF or CRLF + /// -- every line ending the event stream format recognises. + private void putLines(String prefix, String text) { + int start = 0; + int n = text.length(); + for (int i = 0; i <= n; i++) { + char c = i < n ? text.charAt(i) : '\n'; + if (c == '\n' || c == '\r') { + sink.putAscii(prefix); + if (i > start) { + sink.put(' '); + sink.putUtf8(text.substring(start, i)); + } + sink.put('\n'); + if (c == '\r' && i + 1 < n && text.charAt(i + 1) == '\n') { + i++; + } + start = i + 1; + } + } + } + + private void finish() throws IOException { + sink.put('\n'); + write(); + } + + private void write() throws IOException { + out.write(sink.bytes(), 0, sink.length()); + out.flush(); + } + + private static String checkField(String field, String value) { + if (value != null && (value.indexOf('\n') >= 0 || value.indexOf('\r') >= 0)) { + // A line break would end the field and let the rest forge another. + throw new IllegalArgumentException("an event " + field + " cannot contain a line break"); + } + return value; + } + } +}