From cf848240bec14d6e65ae4e02e494529a9089ab27 Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sun, 19 Jul 2026 23:48:55 -0400 Subject: [PATCH 1/4] 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; + +} From 73df15242924b4fe6ddc80c429101260599c79cf Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Wed, 5 Aug 2026 15:51:56 -0400 Subject: [PATCH 2/4] Encapsulate body+streamwriter differences within HttpResponse --- .../common/web/BlueMapResponseModifier.java | 2 +- .../bluemap/common/web/MapRequestHandler.java | 2 +- .../bluemap/common/web/http/HttpResponse.java | 49 +++++++++++++++---- .../web/http/HttpResponseOutputStream.java | 23 +-------- .../web/http/HttpResponseStreamWriter.java | 4 +- 5 files changed, 45 insertions(+), 35 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java b/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java index 25748f2e3..b996a2584 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java @@ -49,7 +49,7 @@ public HttpResponse handle(HttpRequest request) { HttpResponse response = delegate.handle(request); HttpStatusCode status = response.getStatusCode(); - if (status.getCode() >= 400 && response.getBody() != null){ + if (status.getCode() >= 400 && response.hasBody()){ response.setBody(status.getCode() + " - " + status.getMessage() + "\n" + this.serverName); } 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 29c613b66..aaf6badd3 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 @@ -74,7 +74,7 @@ public MapRequestHandler( // attempt to turn off buffering in upstream proxy response.addHeader("X-Accel-Buffering", "no"); - response.setStreamWriter(sseConnections::handleConnection); + response.setBody(sseConnections::handleConnection); return response; }); } 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 fb082f2bc..64ca01bfe 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 @@ -32,28 +32,29 @@ import java.util.LinkedHashMap; import java.util.Map; -@Getter -@Setter @RequiredArgsConstructor public class HttpResponse implements Closeable, HttpHeaderCarrier { - private @NonNull String version = "HTTP/1.1"; - private @NonNull HttpStatusCode statusCode; - private @NonNull @Singular Map headers = new LinkedHashMap<>(); - private @Nullable InputStream body; + private @Getter @Setter @NonNull String version = "HTTP/1.1"; + private @Getter @Setter @NonNull HttpStatusCode statusCode; + private @Getter @Setter @NonNull @Singular Map headers = new LinkedHashMap<>(); /** - * 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. + * The response body. + * + * Can be stored as either an {@link InputStream} or an {@link HttpResponseStreamWriter}. + * The {@link #streamWriter} is used for responses that push data over time (Server-Sent Events). */ + private @Nullable InputStream body; private @Nullable HttpResponseStreamWriter streamWriter; public void setBody(@Nullable InputStream body) { + this.streamWriter = null; this.body = body; } public void setBody(byte[] data) { + this.streamWriter = null; if (data == null) { this.body = null; return; @@ -63,6 +64,7 @@ public void setBody(byte[] data) { } public void setBody(String data) { + this.streamWriter = null; if (data == null) { this.body = null; return; @@ -71,9 +73,38 @@ public void setBody(String data) { setBody(data.getBytes(StandardCharsets.UTF_8)); } + public void setBody(@Nullable HttpResponseStreamWriter streamWriter) { + this.body = null; + this.streamWriter = streamWriter; + } + + public boolean hasBody() { + return body != null || streamWriter != null; + } + @Override public void close() throws IOException { if (body != null) body.close(); } + /** + * Returns {@link #streamWriter} if set, otherwise adapts {@link #body} into a + * {@link HttpResponseStreamWriter} that writes data in chunks. + * Returns {@code null} if neither is set. + */ + @Nullable HttpResponseStreamWriter resolveStreamWriter() { + if (streamWriter != null) return streamWriter; + if (body == null) return null; + + byte[] byteBuffer = new byte[1024]; + return out -> { + while (true) { + int read = body.read(byteBuffer); + if (read == -1) break; + if (read == 0) continue; + out.writeChunk(byteBuffer, 0, read); + } + }; + } + } 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 c06797604..85cdf3627 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 @@ -28,7 +28,6 @@ import java.io.Closeable; import java.io.IOException; -import java.io.InputStream; import java.io.OutputStream; import java.nio.charset.StandardCharsets; @@ -38,15 +37,10 @@ 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); - } + HttpResponseStreamWriter streamWriter = response.resolveStreamWriter(); writeLine(response.getVersion() + " " + statusCode.getCode() + " " + statusCode.getMessage()); @@ -72,21 +66,6 @@ public void write(HttpResponse response) throws IOException { 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; - out.writeChunk(byteBuffer, 0, read); - } - }; - } - private void writeLine() throws IOException { outputStream.write(CRLF); } 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 index a583414db..565c1865f 100644 --- 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 @@ -30,8 +30,8 @@ * 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. + * Used for responses that push data over time (e.g. Server-Sent-Events) rather than producing + * it all up-front. */ @FunctionalInterface public interface HttpResponseStreamWriter { From c2bfb54ec56c6ada1fca96e7aece89ef5c26781f Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Wed, 5 Aug 2026 17:43:24 -0400 Subject: [PATCH 3/4] Always convert HttpResponse.body to a HttpResponseStreamWriter --- .../common/web/BlueMapResponseModifier.java | 2 +- .../bluemap/common/web/http/HttpResponse.java | 75 +++++++------------ .../web/http/HttpResponseOutputStream.java | 2 +- .../web/http/HttpResponseStreamWriter.java | 9 ++- 4 files changed, 35 insertions(+), 53 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java b/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java index b996a2584..25748f2e3 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java @@ -49,7 +49,7 @@ public HttpResponse handle(HttpRequest request) { HttpResponse response = delegate.handle(request); HttpStatusCode status = response.getStatusCode(); - if (status.getCode() >= 400 && response.hasBody()){ + if (status.getCode() >= 400 && response.getBody() != null){ response.setBody(status.getCode() + " - " + status.getMessage() + "\n" + this.serverName); } 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 64ca01bfe..342301719 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 @@ -32,54 +32,30 @@ import java.util.LinkedHashMap; import java.util.Map; +@Getter +@Setter @RequiredArgsConstructor public class HttpResponse implements Closeable, HttpHeaderCarrier { - private @Getter @Setter @NonNull String version = "HTTP/1.1"; - private @Getter @Setter @NonNull HttpStatusCode statusCode; - private @Getter @Setter @NonNull @Singular Map headers = new LinkedHashMap<>(); - - /** - * The response body. - * - * Can be stored as either an {@link InputStream} or an {@link HttpResponseStreamWriter}. - * The {@link #streamWriter} is used for responses that push data over time (Server-Sent Events). - */ - private @Nullable InputStream body; - private @Nullable HttpResponseStreamWriter streamWriter; + private @NonNull String version = "HTTP/1.1"; + private @NonNull HttpStatusCode statusCode; + private @NonNull @Singular Map headers = new LinkedHashMap<>(); + private @Nullable HttpResponseStreamWriter body; public void setBody(@Nullable InputStream body) { - this.streamWriter = null; - this.body = body; + this.body = body == null ? null : asStreamWriter(body); } public void setBody(byte[] data) { - this.streamWriter = null; - if (data == null) { - this.body = null; - return; - } - - setBody(new ByteArrayInputStream(data)); + setBody(data == null ? null : new ByteArrayInputStream(data)); } public void setBody(String data) { - this.streamWriter = null; - if (data == null) { - this.body = null; - return; - } - - setBody(data.getBytes(StandardCharsets.UTF_8)); + setBody(data == null ? null : data.getBytes(StandardCharsets.UTF_8)); } public void setBody(@Nullable HttpResponseStreamWriter streamWriter) { - this.body = null; - this.streamWriter = streamWriter; - } - - public boolean hasBody() { - return body != null || streamWriter != null; + this.body = streamWriter; } @Override @@ -88,21 +64,26 @@ public void close() throws IOException { } /** - * Returns {@link #streamWriter} if set, otherwise adapts {@link #body} into a - * {@link HttpResponseStreamWriter} that writes data in chunks. - * Returns {@code null} if neither is set. + * Adapt an {@link InputStream} body into a {@link HttpResponseStreamWriter} + * that writes the data to the client in chunks. */ - @Nullable HttpResponseStreamWriter resolveStreamWriter() { - if (streamWriter != null) return streamWriter; - if (body == null) return null; + private static HttpResponseStreamWriter asStreamWriter(InputStream body) { + return new HttpResponseStreamWriter() { + private final byte[] byteBuffer = new byte[1024]; + + @Override + public void write(ChunkedOutputStream out) throws IOException { + while (true) { + int read = body.read(byteBuffer); + if (read == -1) break; + if (read == 0) continue; + out.writeChunk(byteBuffer, 0, read); + } + } - byte[] byteBuffer = new byte[1024]; - return out -> { - while (true) { - int read = body.read(byteBuffer); - if (read == -1) break; - if (read == 0) continue; - out.writeChunk(byteBuffer, 0, read); + @Override + public void close() throws IOException { + body.close(); } }; } 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 85cdf3627..82215ac65 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 @@ -40,7 +40,7 @@ public class HttpResponseOutputStream implements Closeable { public void write(HttpResponse response) throws IOException { HttpStatusCode statusCode = response.getStatusCode(); - HttpResponseStreamWriter streamWriter = response.resolveStreamWriter(); + HttpResponseStreamWriter streamWriter = response.getBody(); writeLine(response.getVersion() + " " + statusCode.getCode() + " " + statusCode.getMessage()); 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 index 565c1865f..693216a44 100644 --- 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 @@ -24,18 +24,19 @@ */ package de.bluecolored.bluemap.common.web.http; +import java.io.Closeable; 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 for responses that push data over time (e.g. Server-Sent-Events) rather than producing - * it all up-front. */ @FunctionalInterface -public interface HttpResponseStreamWriter { +public interface HttpResponseStreamWriter extends Closeable { void write(ChunkedOutputStream out) throws IOException; + @Override + default void close() throws IOException {} + } From 2cd18063247bc84379ec3981795dfc0382e392ea Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Wed, 5 Aug 2026 18:02:59 -0400 Subject: [PATCH 4/4] Add a soft size limit to how much data ChunkedOutputStream will buffer --- .../bluemap/common/web/http/ChunkedOutputStream.java | 3 +++ 1 file changed, 3 insertions(+) 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 index b1a5840b8..6ea5386cb 100644 --- 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 @@ -43,6 +43,7 @@ */ public class ChunkedOutputStream extends OutputStream { + private static final int AUTO_FLUSH_CHUNK_SIZE = 1024; private static final byte[] CRLF = "\r\n".getBytes(StandardCharsets.UTF_8); private final OutputStream out; @@ -57,12 +58,14 @@ public ChunkedOutputStream(OutputStream out) { public void write(int b) throws IOException { ensureOpen(); buffer.write(b); + if (buffer.size() >= AUTO_FLUSH_CHUNK_SIZE) endChunk(); } @Override public void write(byte[] b, int off, int len) throws IOException { ensureOpen(); buffer.write(b, off, len); + if (buffer.size() >= AUTO_FLUSH_CHUNK_SIZE) endChunk(); } /**