From d9af9139602dfadd4a0b285355c5c23a9b1bc62d Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sun, 19 Jul 2026 23:48:55 -0400 Subject: [PATCH] Optimize how SSE data is written The `SseConnection` now writes queued events directly to the response's output stream from the connection-handling thread itself. Previously each SSE connection has its own virtual thread that would create and push the events into a pipe for the actual connection-handling thread to read from. Per SSE connection, this new method saves the creation of a virtual thread, a `PipedOutputStream`/`PipedInputStream` pair and their internal buffer, and the overhead of having to push all the SSE data across a thread boundary. Additionally, because the `SseConnection` now handles writing the data to the output stream itself and can handle flushing it as needed, the `HttpResponseOutputStream` no longer needs to flush the output stream after every buffer read just in case the data it was sending was part of an SSE stream. To support this change, the `HttpResponseStreamWriter` interface was added to allow `HttpResponse` bodies to be streamed incrementally and dynamically chunked using the new `ChunkedOutputStream` rather than only being able to be read fixed-size chunks from an `InputStream`. Note that because the pipe buffer is now gone, event buffering in the `SseConnection` occurs entirely in the queue. To compensate for this, the `QUEUE_CAPACITY` was increased from 16 to 64. --- .../bluemap/common/web/MapRequestHandler.java | 7 +- .../bluemap/common/web/SseConnection.java | 90 +++++------- .../common/web/SseConnectionManager.java | 14 +- .../common/web/http/ChunkedOutputStream.java | 133 ++++++++++++++++++ .../bluemap/common/web/http/HttpResponse.java | 7 + .../web/http/HttpResponseOutputStream.java | 35 +++-- .../web/http/HttpResponseStreamWriter.java | 41 ++++++ 7 files changed, 247 insertions(+), 80 deletions(-) create mode 100644 common/src/main/java/de/bluecolored/bluemap/common/web/http/ChunkedOutputStream.java create mode 100644 common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseStreamWriter.java diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/MapRequestHandler.java b/common/src/main/java/de/bluecolored/bluemap/common/web/MapRequestHandler.java index f0e4603c1..29c613b66 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/MapRequestHandler.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/MapRequestHandler.java @@ -32,7 +32,6 @@ import com.flowpowered.math.vector.Vector2i; -import java.io.IOException; import java.util.function.Consumer; import java.util.function.Supplier; @@ -75,11 +74,7 @@ public MapRequestHandler( // attempt to turn off buffering in upstream proxy response.addHeader("X-Accel-Buffering", "no"); - try { - response.setBody(sseConnections.openConnection()); - } catch (IOException e) { - return new HttpResponse(HttpStatusCode.INTERNAL_SERVER_ERROR); - } + response.setStreamWriter(sseConnections::handleConnection); return response; }); } diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java index e211cb795..af805dc0a 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java @@ -26,55 +26,30 @@ import java.io.Closeable; import java.io.IOException; -import java.io.InputStream; -import java.io.PipedInputStream; -import java.io.PipedOutputStream; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; -import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; -import de.bluecolored.bluemap.core.util.stream.OnCloseInputStream; import lombok.SneakyThrows; /** * Represents a single Server-Sent Events (SSE) connection. *

- * Read the events from the {@link PipedInputStream} returned from {@link #getInputStream()}. - * Reading from the stream will block until a new event is delivered to it. - *

- * Events are queued via {@link #enqueue(String, String)} and delivered via a virtual thread - * owned by this connection so a slow client only blocks its own delivery. + * Events can be queued via {@link #enqueue(String, String)} without blocking. + * Call {@link #run(OutputStream)} on the thread that owns the connection's output-stream (e.g. + * the HTTP connection's thread) to deliver queued events to it. This will block the calling thread + * until the connection is closed. */ public class SseConnection implements Closeable { - private static final int PIPE_BUFFER_SIZE = 1024; + // how many messages can be queued up for sending before being dropped + private static final int QUEUE_CAPACITY = 64; - // how many messages can be queued up for sending (in addition to the above buffer) - // before being dropped - private static final int QUEUE_CAPACITY = 16; - - private final PipedOutputStream pipeOut; - private final InputStream pipeIn; private final BlockingQueue queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); - private final Thread sendThread; private volatile boolean closed = false; private volatile Runnable onClose; - - public SseConnection() throws IOException { - // add a hook to the pipe to close the conneciton if the stream is closed - this.pipeOut = new PipedOutputStream(); - this.pipeIn = new OnCloseInputStream(new PipedInputStream(pipeOut, PIPE_BUFFER_SIZE), SseConnection.this); - - this.sendThread = Thread.ofVirtual().name("BlueMap-SSE-send").start(this::sendLoop); - } - - /** - * Returns an {@link InputStream} to read events from. - * Closing it also closes this connection. - */ - public InputStream getInputStream() { - return pipeIn; - } + private volatile Thread runningThread; public boolean isClosed() { return closed; @@ -104,44 +79,51 @@ public void enqueue(String eventType, String data) { } } - private void sendLoop() { + /** + * Delivers queued events directly to {@code out}, blocking the calling thread until this + * connection is closed either explicitly via {@link #close()}, or because writing to + * {@code out} fails (happens if the client disconnects). + */ + public void run(OutputStream out) throws IOException { + runningThread = Thread.currentThread(); + String[] event; try { while (!closed) { - String[] event = queue.take(); - send(event[0], event[1]); + try { + event = queue.take(); + } catch (InterruptedException _) { + runningThread.interrupt(); + break; + } + send(out, event[0], event[1]); } - } catch (InterruptedException | IOException ignored) {} + } finally { + close(); + } } @SneakyThrows(IOException.class) // allows using this function in the forEach below - private void writeLine(String line){ - pipeOut.write((line + "\n").getBytes(StandardCharsets.UTF_8)); + private void writeLine(OutputStream out, String line) { + out.write((line + "\n").getBytes(StandardCharsets.UTF_8)); } /** * Write one SSE event with optional data to the stream and flush it. * - * @throws IOException if the connection is closed or the client has disconnected + * @throws IOException if the client has disconnected */ - private synchronized void send(String eventType, String data) throws IOException { - if (closed) throw new IOException("SSE connection is closed"); - try { - writeLine("event: " + eventType); - data.lines().forEach(l -> writeLine("data: " + l)); - pipeOut.write('\n'); - pipeOut.flush(); - } catch (IOException e) { - close(); - throw e; - } + private void send(OutputStream out, String eventType, String data) throws IOException { + writeLine(out, "event: " + eventType); + data.lines().forEach(l -> writeLine(out, "data: " + l)); + out.write('\n'); + out.flush(); } @Override public synchronized void close() { if (closed) return; closed = true; - sendThread.interrupt(); - try { pipeOut.close(); } catch (IOException ignored) {} + if (runningThread != null) runningThread.interrupt(); if (onClose != null) onClose.run(); } diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java index 4f462adb5..910e82bd4 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java @@ -26,7 +26,7 @@ import java.io.Closeable; import java.io.IOException; -import java.io.InputStream; +import java.io.OutputStream; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; @@ -54,15 +54,15 @@ public void removeHasConnectionsListener(Consumer listener) { } /** - * Creates a new {@link SseConnection}, registers it, and returns an {@link InputStream} suitable - * for use as an HTTP response body. When the stream is closed (either because the client - * disconnected or the server closed the connection), the connection is automatically removed - * from this manager. + * Creates a new {@link SseConnection}, registers it, and delivers events to {@code out} until + * the connection closes (either because the client disconnected or the server closed the + * connection), blocking the calling thread for that whole time. The connection is + * automatically removed from this manager once it closes. */ - public InputStream openConnection() throws IOException { + public void handleConnection(OutputStream out) throws IOException { SseConnection connection = new SseConnection(); add(connection); - return connection.getInputStream(); + connection.run(out); } public void add(SseConnection connection) { diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/http/ChunkedOutputStream.java b/common/src/main/java/de/bluecolored/bluemap/common/web/http/ChunkedOutputStream.java new file mode 100644 index 000000000..b1a5840b8 --- /dev/null +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/http/ChunkedOutputStream.java @@ -0,0 +1,133 @@ +/* + * This file is part of BlueMap, licensed under the MIT License (MIT). + * + * Copyright (c) Blue (Lukas Rieger) + * Copyright (c) contributors + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package de.bluecolored.bluemap.common.web.http; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; + +/** + * Wraps an {@link OutputStream}, buffering writes and framing them as HTTP/1.1 chunks. + *

+ * {@link #endChunk()} ends the current chunk without flushing the wrapped stream. + * {@link #flush()} ends the current chunk *and* flushes the wrapped stream. + *

+ * Closing this stream ends the current chunk and writes the terminating zero-length chunk, but + * doesn't close the wrapped stream since it's expected to outlive an individual chunked response. + *

+ * Any write made after this stream has been closed throws an {@link IOException} to avoid bytes + * being written to the wrapped stream outside the required chunk framing. + */ +public class ChunkedOutputStream extends OutputStream { + + private static final byte[] CRLF = "\r\n".getBytes(StandardCharsets.UTF_8); + + private final OutputStream out; + private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + private boolean closed = false; + + public ChunkedOutputStream(OutputStream out) { + this.out = out; + } + + @Override + public void write(int b) throws IOException { + ensureOpen(); + buffer.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + ensureOpen(); + buffer.write(b, off, len); + } + + /** + * Writes out any currently buffered bytes as one HTTP chunk. + */ + public void endChunk() throws IOException { + ensureOpen(); + if (buffer.size() > 0) { + writeChunkHeader(buffer.size()); + buffer.writeTo(out); + out.write(CRLF); + buffer.reset(); + } + } + + /** + * Writes {@code len} bytes from {@code b} starting at {@code off} as single chunk. + * This avoids the buffering overhead incurred by using {@code write(...)}. + *

+ * Any currently buffered bytes are written with {@link #endChunk()} first. + */ + public void writeChunk(byte[] b, int off, int len) throws IOException { + endChunk(); + if (len > 0) { + writeChunkHeader(len); + out.write(b, off, len); + out.write(CRLF); + } + } + + private void writeChunkHeader(int len) throws IOException { + out.write(Integer.toHexString(len).getBytes(StandardCharsets.UTF_8)); + out.write(CRLF); + } + + /** + * Ends the current chunk and flushes the wrapped stream to push all the buffered + * data to the client. + */ + @Override + public void flush() throws IOException { + endChunk(); + out.flush(); + } + + /** + * Ends the current chunk, writes the terminating zero-length chunk, and flushes the + * wrapped stream (without closing it). + */ + @Override + public void close() throws IOException { + if (closed) return; + endChunk(); + closed = true; + out.write('0'); + out.write(CRLF); + out.write(CRLF); + out.flush(); + } + + /** + * @throws IOException if this stream has already been closed. + */ + private void ensureOpen() throws IOException { + if (closed) throw new IOException("stream closed"); + } + +} diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponse.java b/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponse.java index 610ad3795..fb082f2bc 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponse.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponse.java @@ -42,6 +42,13 @@ public class HttpResponse implements Closeable, HttpHeaderCarrier { private @NonNull @Singular Map headers = new LinkedHashMap<>(); private @Nullable InputStream body; + /** + * If set, takes over writing this response's body directly to the connection's output-stream + * instead of reading it from {@link #body}. + * Used for responses that push data over time like Server-Sent Events. + */ + private @Nullable HttpResponseStreamWriter streamWriter; + public void setBody(@Nullable InputStream body) { this.body = body; } diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseOutputStream.java b/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseOutputStream.java index d1b053e00..c06797604 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseOutputStream.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseOutputStream.java @@ -38,17 +38,20 @@ public class HttpResponseOutputStream implements Closeable { private static final byte[] CRLF = "\r\n".getBytes(StandardCharsets.UTF_8); private final OutputStream outputStream; - private final byte[] byteBuffer = new byte[1024]; public void write(HttpResponse response) throws IOException { HttpStatusCode statusCode = response.getStatusCode(); InputStream body = response.getBody(); + HttpResponseStreamWriter streamWriter = response.getStreamWriter(); + if (streamWriter == null && body != null) { + streamWriter = asStreamWriter(body); + } writeLine(response.getVersion() + " " + statusCode.getCode() + " " + statusCode.getMessage()); // headers - if (body != null) { + if (streamWriter != null) { response.addHeader("Transfer-Encoding","chunked"); } else { response.addHeader("Content-Length", "0"); @@ -57,25 +60,31 @@ public void write(HttpResponse response) throws IOException { writeLine(header.getKey() + ": " + header.getValue()); } writeLine(); + outputStream.flush(); // ensure headers are always immediately pushed to the client // body - if (body != null) { + if (streamWriter != null) { + try (ChunkedOutputStream chunkedOut = new ChunkedOutputStream(outputStream)){ + streamWriter.write(chunkedOut); + } + } + outputStream.flush(); + } + + /** + * Adapt an {@link InputStream} body into a {@link HttpResponseStreamWriter} + * that writes the data to the client in chunks. + */ + private HttpResponseStreamWriter asStreamWriter(InputStream body) { + return out -> { while (true) { int read = body.read(byteBuffer); if (read == -1) break; if (read == 0) continue; - writeLine(Integer.toHexString(read)); - outputStream.write(byteBuffer, 0, read); - writeLine(); - outputStream.flush(); // prevent SSE from being buffered + out.writeChunk(byteBuffer, 0, read); } - - writeLine(Integer.toHexString(0)); - writeLine(); - } - - outputStream.flush(); + }; } private void writeLine() throws IOException { diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseStreamWriter.java b/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseStreamWriter.java new file mode 100644 index 000000000..a583414db --- /dev/null +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpResponseStreamWriter.java @@ -0,0 +1,41 @@ +/* + * This file is part of BlueMap, licensed under the MIT License (MIT). + * + * Copyright (c) Blue (Lukas Rieger) + * Copyright (c) contributors + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package de.bluecolored.bluemap.common.web.http; + +import java.io.IOException; + +/** + * Writes a {@link HttpResponse}'s body directly to the connection's output-stream, taking over + * writing to (and blocking) the calling thread until the body is fully written. + *

+ * Used instead of {@link HttpResponse#getBody()} for responses that push data over time + * (e.g. Server-Sent-Events) rather than producing it all up-front. + */ +@FunctionalInterface +public interface HttpResponseStreamWriter { + + void write(ChunkedOutputStream out) throws IOException; + +}