From 0d2f2c95b2f9e28588a5dac614a1c39e265cff42 Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Fri, 15 May 2026 01:08:19 -0700 Subject: [PATCH 01/12] Use Server-sent events to add live updates for files, players, and markers SSE Implementation: - A `SseConnectionManager` is created per-map in the `MapRequestHandler`. - Whenever a client connects to the `live/sse` endpoint, a new `SseConnection` is created for it and is added to the `SseConnectionManager`. - The `SseConnectionManager` receives requests to push updates and sends them to all known `SseConnections`. If sending fails, the connection is cleaned up and removed. - Sending happens on a background thread so normal server operations are not blocked on clients picking up events. Backend events: - Tile updates are event-based: when a new tile is rendered an event is emitted which causes the tile's X+Y and lod to be broadcasted to all connected clients. - Players and other markers are still polling-based, but the polling is done server-side (and only once per endpoint) and connected clients are notified only when the data changes. - Polling was implemented by replacing the `CachedRateLimitDataSupplier` with a `LiveDataSupplierBroadcaster`. This `Supplier` wrapper can be used in the same way where requests to `get()` it's data are rate-limited and cached. However, it also runs a background thread that (when listeners are added) will automatically poll the supplier and send events when the data changes. Frontend changes: - On page load, an `EventSource` connects to the `/live/sse` endpoint and watches for events. - "tile" updates cause the specified tiles to be force-reloaded - "player" updates are fed directly into the existing `playerMarkerManager` - "marker" updates are fed directly into the existing `markerFileManager` - Since this takes the place of the marker managers polling, the marker managers were updated to be able to be paused, and are started in a paused state. - In order to ensure that markers are still updated if the SSE endpoint can't be contacted, the existing polling marker managers will be unpaused if the SSE connection fails to connect or disconnects. If the SSE connection returns, the marker managers will be paused again. - Tile force-reloading is handled by removing the specified tile from the `revalidatedUrls` cache, the calling `load` on it. This forces the normal tile loading mechanism to update the tile, reloading it. --- .../bluemap/common/plugin/Plugin.java | 2 +- .../web/CachedRateLimitDataSupplier.java | 64 ------- .../web/LiveDataSupplierBroadcaster.java | 129 ++++++++++++++ .../bluemap/common/web/MapRequestHandler.java | 73 +++++++- .../bluemap/common/web/SseConnection.java | 94 ++++++++++ .../common/web/SseConnectionManager.java | 166 ++++++++++++++++++ .../web/http/HttpResponseOutputStream.java | 1 + common/webapp/src/js/BlueMapApp.js | 63 ++++++- common/webapp/src/js/map/LowresTileLoader.js | 5 +- common/webapp/src/js/map/Tile.js | 4 +- common/webapp/src/js/map/TileLoader.js | 5 +- common/webapp/src/js/markers/MarkerManager.js | 38 +++- .../src/js/markers/NormalMarkerManager.js | 4 +- .../src/js/markers/PlayerMarkerManager.js | 4 +- .../core/map/hires/HiresModelManager.java | 16 ++ .../bluemap/core/map/lowres/LowresLayer.java | 16 ++ .../core/map/lowres/LowresTileManager.java | 14 ++ 17 files changed, 606 insertions(+), 92 deletions(-) delete mode 100644 common/src/main/java/de/bluecolored/bluemap/common/web/CachedRateLimitDataSupplier.java create mode 100644 common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java create mode 100644 common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java create mode 100644 common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java diff --git a/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java b/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java index a61e3721d..876808001 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java @@ -234,7 +234,7 @@ private void load(@Nullable ResourcePack preloadedResourcePack) throws IOExcepti null; LiveMarkersDataSupplier liveMarkersDataSupplier = new LiveMarkersDataSupplier(map.getMarkerSets()); - mapRequestHandler = new MapRequestHandler(map.getStorage(), livePlayersDataSupplier, liveMarkersDataSupplier); + mapRequestHandler = new MapRequestHandler(map, livePlayersDataSupplier, liveMarkersDataSupplier); } else { Storage storage = blueMap.getOrLoadStorage(mapConfig.getStorage()); mapRequestHandler = new MapRequestHandler(storage.map(id)); diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/CachedRateLimitDataSupplier.java b/common/src/main/java/de/bluecolored/bluemap/common/web/CachedRateLimitDataSupplier.java deleted file mode 100644 index d14e0798f..000000000 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/CachedRateLimitDataSupplier.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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; - -import java.util.concurrent.locks.ReentrantLock; -import java.util.function.Supplier; - -public class CachedRateLimitDataSupplier implements Supplier { - - private final ReentrantLock lock = new ReentrantLock(); - - private final Supplier delegate; - private final long rateLimitMillis; - - private long updateTime = -1; - private String data = null; - - public CachedRateLimitDataSupplier(Supplier delegate, long rateLimitMillis) { - this.delegate = delegate; - this.rateLimitMillis = rateLimitMillis; - } - - @Override - public String get() { - update(); - return data; - } - - protected void update() { - if (lock.tryLock()) { - try { - long now = System.currentTimeMillis(); - if (data != null && now < updateTime + this.rateLimitMillis) return; - this.data = delegate.get(); - this.updateTime = now; - } finally { - lock.unlock(); - } - } - } - -} diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java b/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java new file mode 100644 index 000000000..3810239fe --- /dev/null +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java @@ -0,0 +1,129 @@ +/* + * 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; + +import java.io.Closeable; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; +import java.util.function.Supplier; + +/** + * Polls a {@link Supplier} on a background thread and notifies registered listeners + * whenever the returned value changes. + * + * Call {@link #update()} to get the current value (rate-limited by the polling rate) + * Call {@link #close()} to stop the background thread. + */ +public class LiveDataSupplierBroadcaster implements Supplier, Closeable { + + private final Supplier dataSupplier; + private final long pollIntervalMillis; + private final Set> listeners = ConcurrentHashMap.newKeySet(); + private final Thread pollThread; + private volatile boolean closed = false; + private long lastUpdate = -1; + private T data = null; + + public LiveDataSupplierBroadcaster(Supplier dataSupplier, long pollIntervalMillis) { + this.dataSupplier = dataSupplier; + this.pollIntervalMillis = pollIntervalMillis; + this.pollThread = new Thread(this::pollLoop, String.format("%sPoller", dataSupplier.getClass().getName())); + this.pollThread.setDaemon(true); + this.pollThread.start(); + } + + public synchronized void addUpdateListener(Consumer listener) { + listeners.add(listener); + notifyAll(); + } + + public void removeUpdateListener(Consumer listener) { + listeners.remove(listener); + } + + private void pollLoop() { + while (!closed) { + // suspend polling until at least one listener is registered + synchronized (this) { + try { + while (listeners.isEmpty() && !closed){ + wait(); + } + } catch (InterruptedException ignored){ + break; + } + } + if (closed) break; + update(); + + try { + Thread.sleep(pollIntervalMillis); + } catch (InterruptedException ignored) { + break; + } + } + } + + /** + * Ensure the data is up to date and return it. + * + * Note that this will only get new data from the supplier if the current + * cached data is stale (according to {@code pollIntervalMillis}). + */ + @Override + public T get() { + update(); + return this.data; + } + + private void update() { + synchronized (this) { + long now = System.currentTimeMillis(); + if (this.data != null && now < this.lastUpdate + this.pollIntervalMillis) return; + this.lastUpdate = now; + + T newdata = dataSupplier.get(); + if (newdata == null || newdata.equals(this.data)) return; + + // new data, update listeners + this.data = newdata; + } + for (Consumer listener : listeners) { + listener.accept(this.data); + } + } + + /** + * Stops the background polling thread. + */ + @Override + public synchronized void close() { + closed = true; + notifyAll(); + pollThread.interrupt(); + } + +} 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 16122e429..42a085a04 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 @@ -24,13 +24,36 @@ */ package de.bluecolored.bluemap.common.web; +import de.bluecolored.bluemap.common.web.http.HttpRequestHandler; +import de.bluecolored.bluemap.common.web.http.HttpResponse; +import de.bluecolored.bluemap.common.web.http.HttpStatusCode; +import de.bluecolored.bluemap.core.map.BmMap; import de.bluecolored.bluemap.core.storage.MapStorage; import org.jetbrains.annotations.Nullable; +import com.flowpowered.math.vector.Vector2i; + +import java.io.IOException; +import java.util.function.Consumer; import java.util.function.Supplier; public class MapRequestHandler extends RoutingRequestHandler { + private final SseConnectionManager sseConnections = new SseConnectionManager(); + + public MapRequestHandler( + BmMap map, + @Nullable Supplier livePlayersDataSupplier, + @Nullable Supplier liveMarkerDataSupplier + ) { + this(map.getStorage(), livePlayersDataSupplier, liveMarkerDataSupplier); + + // only register the handler for map updates if we're given the actual map + // instance from the plugin (ie. not running standalone) + map.getHiresModelManager().addTileUpdateListener(tile -> onTileUpdate(tile, 0)); + map.getLowresTileManager().addTileUpdateListener((tile, lod) -> onTileUpdate(tile, lod)); + } + public MapRequestHandler(MapStorage mapStorage) { this(mapStorage, null, null); } @@ -42,17 +65,55 @@ public MapRequestHandler( ) { register(".*", new MapStorageRequestHandler(mapStorage)); + register("live/sse", "", (HttpRequestHandler) request -> { + HttpResponse response = new HttpResponse(HttpStatusCode.OK); + response.addHeader("Content-Type", "text/event-stream"); + response.addHeader("Cache-Control", "no-cache"); + try { + response.setBody(sseConnections.openConnection()); + } catch (IOException e) { + return new HttpResponse(HttpStatusCode.INTERNAL_SERVER_ERROR); + } + return response; + }); + if (livePlayersDataSupplier != null) { - register("live/players\\.json", "", new JsonDataRequestHandler( - new CachedRateLimitDataSupplier(livePlayersDataSupplier,1000) - )); + LiveDataSupplierBroadcaster playerDataBroadcaster = new LiveDataSupplierBroadcaster<>(livePlayersDataSupplier, 1000); + registerSseCallback(playerDataBroadcaster, this::onPlayerUpdate); + register("live/players\\.json", "", new JsonDataRequestHandler(playerDataBroadcaster)); } if (liveMarkerDataSupplier != null) { - register("live/markers\\.json", "", new JsonDataRequestHandler( - new CachedRateLimitDataSupplier(liveMarkerDataSupplier,10000) - )); + LiveDataSupplierBroadcastermarkerDataBroadcaster = new LiveDataSupplierBroadcaster<>(liveMarkerDataSupplier, 10000); + registerSseCallback(markerDataBroadcaster, this::onMarkerUpdate); + register("live/markers\\.json", "", new JsonDataRequestHandler(markerDataBroadcaster)); } } + /** + * Helper function to subscribe to updates from a broadcaster (forcing it to auto-refresh) + * only if the SSE manager has a connection. + */ + private void registerSseCallback(LiveDataSupplierBroadcaster broadcaster, Consumer callback){ + sseConnections.addHasConnectionsListener(hasConnections -> { + if (hasConnections) { + broadcaster.addUpdateListener(callback); + } else { + broadcaster.removeUpdateListener(callback); + } + }); + } + + private void onTileUpdate(Vector2i tile, int lod) { + // since the data is all ints there's no escaping issues so just build the JSON the hacky fast way + sseConnections.broadcast("tile", "{\"x\":" + tile.getX() + ",\"y\":" + tile.getY() + ",\"lod\":" + lod + "}"); + } + + private void onPlayerUpdate(String data) { + sseConnections.broadcast("player", data); + } + + private void onMarkerUpdate(String data) { + sseConnections.broadcast("marker", data); + } } 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 new file mode 100644 index 000000000..d8e08e5dc --- /dev/null +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java @@ -0,0 +1,94 @@ +/* + * 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; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.nio.charset.StandardCharsets; + +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 sent to it via {@link #send(String, String)}. + * Sending an event will flush the stream, ensuring that all events can be read immediately. + */ +public class SseConnection implements Closeable { + + private static final int PIPE_BUFFER_SIZE = 4096; + + private final PipedOutputStream pipeOut; + private final PipedInputStream pipeIn; + private volatile boolean closed = false; + + public SseConnection() throws IOException { + this.pipeOut = new PipedOutputStream(); + this.pipeIn = new PipedInputStream(pipeOut, PIPE_BUFFER_SIZE); + } + + public InputStream getInputStream() { + return pipeIn; + } + + public boolean isClosed() { + return closed; + } + + @SneakyThrows(IOException.class) // allows using this function in the forEach below + private void writeLine(String line){ + pipeOut.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 + */ + public 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; + } + } + + @Override + public synchronized void close() { + if (closed) return; + closed = true; + try { pipeOut.close(); } catch (IOException ignored) {} + } + +} 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 new file mode 100644 index 000000000..cf8d27920 --- /dev/null +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java @@ -0,0 +1,166 @@ +/* + * 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; + +import java.io.Closeable; +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.function.Consumer; + +/** + * Tracks active {@link SseConnection}s and provides queued broadcast delivery. + */ +public class SseConnectionManager implements Closeable { + + private final Set connections = ConcurrentHashMap.newKeySet(); + private final LinkedBlockingQueue broadcastQueue = new LinkedBlockingQueue<>(); + private final Thread broadcastThread; + private volatile boolean closed; + + // allow objects to listen for when the connection count transitions between 0 and non-0 + private final Set> hasConnectionsListeners = ConcurrentHashMap.newKeySet(); + + public void addHasConnectionsListener(Consumer listener) { hasConnectionsListeners.add(listener); } + public void removeHasConnectionsListener(Consumer listener) { hasConnectionsListeners.remove(listener); } + + public SseConnectionManager() { + this.broadcastThread = new Thread(this::broadcastLoop, "bluemap-sse-broadcast"); + this.broadcastThread.setDaemon(true); + this.broadcastThread.start(); + } + + /** + * 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. + */ + public InputStream openConnection() throws IOException { + SseConnection connection = new SseConnection(); + add(connection); + return new FilterInputStream(connection.getInputStream()) { + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + connection.close(); + remove(connection); + } + } + }; + } + + public void add(SseConnection connection) { + boolean fire; + synchronized (this) { + fire = connections.isEmpty(); + fire = connections.add(connection) && fire; + } + if (fire) notifyHasConnections(true); + } + + public void remove(SseConnection connection) { + boolean fire; + synchronized (this) { + fire = connections.remove(connection) && connections.isEmpty(); + } + if (fire) notifyHasConnections(false); + } + + /** + * Queues an SSE event to be sent to all live connections on the background broadcast thread. + * Returns immediately without blocking. + * + * @param eventType the SSE event type + * @param data the event data payload + */ + public void broadcast(String eventType, String data) { + if (closed) return; + broadcastQueue.offer(new String[]{eventType, data}); + } + + private void broadcastLoop() { + try { + while (!closed) { + String[] event = broadcastQueue.take(); + broadcastSync(event[0], event[1]); + } + } catch (InterruptedException ignored) {} + } + + /** + * Sends an SSE event to all live connections synchronously. + * Dead or broken connections are removed automatically. + */ + private void broadcastSync(String eventType, String data) { + if (connections.isEmpty()) return; + List toRemove = new ArrayList<>(); + for (SseConnection conn : connections) { + try { + conn.send(eventType, data); + } catch (IOException ignored) {} + if (conn.isClosed()) { + toRemove.add(conn); + } + } + if (!toRemove.isEmpty()) { + boolean fire; + synchronized (this) { + fire = connections.removeAll(toRemove) && connections.isEmpty(); + } + if (fire) notifyHasConnections(false); + } + } + + /** + * Closes all registered connections, clears the registry, and stops the broadcast thread. + */ + @Override + public void close() { + closed = true; + broadcastThread.interrupt(); + boolean fire; + synchronized (this) { + fire = !connections.isEmpty(); + for (SseConnection conn : connections) { + conn.close(); + } + connections.clear(); + } + if (fire) notifyHasConnections(false); + } + + private void notifyHasConnections(boolean hasConnections) { + hasConnectionsListeners.forEach(listener -> listener.accept(hasConnections)); + } + +} 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 ea1abb715..d1b053e00 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 @@ -68,6 +68,7 @@ public void write(HttpResponse response) throws IOException { writeLine(Integer.toHexString(read)); outputStream.write(byteBuffer, 0, read); writeLine(); + outputStream.flush(); // prevent SSE from being buffered } writeLine(Integer.toHexString(0)); diff --git a/common/webapp/src/js/BlueMapApp.js b/common/webapp/src/js/BlueMapApp.js index ba3fbc685..6017dcf44 100644 --- a/common/webapp/src/js/BlueMapApp.js +++ b/common/webapp/src/js/BlueMapApp.js @@ -28,7 +28,7 @@ import {MapControls} from "./controls/map/MapControls"; import {FreeFlightControls} from "./controls/freeflight/FreeFlightControls"; import {MathUtils, Vector3} from "three"; import {Map as BlueMapMap} from "./map/Map"; -import {alert, animate, EasingFunctions} from "./util/Utils"; +import {alert, animate, EasingFunctions, hashTile} from "./util/Utils"; import {MainMenu} from "./MainMenu"; import {PopupMarker} from "./PopupMarker"; import {MarkerSet} from "./markers/MarkerSet"; @@ -57,6 +57,9 @@ export class BlueMapApp { /** @type {NormalMarkerManager} */ this.markerFileManager = null; + /** @type {EventSource} */ + this.mapEventSource = null; + /** @type {{ * version: string, * useCookies: boolean, @@ -108,6 +111,11 @@ export class BlueMapApp { debug: false }); + // close SSE connection when the page is closed + window.addEventListener("beforeunload", () => { + if (this.mapEventSource) this.mapEventSource.close(); + }); + // init this.updateControlsSettings(); this.initGeneralEvents(); @@ -265,7 +273,8 @@ export class BlueMapApp { await Promise.all([ this.initPlayerMarkerManager(), - this.initMarkerFileManager() + this.initMarkerFileManager(), + this.initEventSource(), ]); } @@ -400,6 +409,51 @@ export class BlueMapApp { }); } + initEventSource(){ + if (this.mapEventSource) { + this.mapEventSource.close(); + } + + const map = this.mapViewer.map; + if (!map) return; + + this.mapEventSource = new EventSource(map.data.liveDataRoot + "/live/sse"); + this.mapEventSource.addEventListener("error", () => { + alert(this.events, "SSE event source error - enabling polling", "debug"); + this.playerMarkerManager.resumeAutoUpdates(); + this.markerFileManager.resumeAutoUpdates(); + }); + this.mapEventSource.addEventListener("open", () => { + alert(this.events, "Connected to SSE event source - disabling polling", "debug"); + this.playerMarkerManager.pauseAutoUpdates(); + this.markerFileManager.pauseAutoUpdates(); + }); + + this.mapEventSource.addEventListener("tile", ({ data }) => { + alert(this.events, `tile update: ${data}`, "debug"); + const parsed = JSON.parse(data); + + const mgr = parsed.lod > 0 ? map.lowresTileManager[parsed.lod - 1] : map.hiresTileManager; + if (!mgr.unloaded) { + const tilehash = hashTile(parsed.x, parsed.y); + const tile = mgr.tiles.get(tilehash); + if (tile && !tile.loading) { + tile.load(mgr.tileLoader, true); + } + } + }); + + this.mapEventSource.addEventListener("player", ({ data }) => { + alert(this.events, `player update: ${data}`, "debug"); + this.playerMarkerManager.updateFromData(JSON.parse(data)); + }); + + this.mapEventSource.addEventListener("marker", ({ data }) => { + alert(this.events, `marker update: ${data}`, "debug"); + this.markerFileManager.updateFromData(JSON.parse(data)); + }); + } + initPlayerMarkerManager() { if (this.playerMarkerManager) this.playerMarkerManager.dispose() @@ -411,7 +465,8 @@ export class BlueMapApp { this.mapViewer.markers, map.data.liveDataRoot + "/live/players.json", map.data.mapDataRoot + "/assets/playerheads/", - this.events + this.events, + true ); this.playerMarkerManager.setAutoUpdateInterval(0); return this.playerMarkerManager.update() @@ -431,7 +486,7 @@ export class BlueMapApp { const map = this.mapViewer.map; if (!map) return; - this.markerFileManager = new NormalMarkerManager(this.mapViewer.markers, map.data.liveDataRoot + "/live/markers.json", this.events); + this.markerFileManager = new NormalMarkerManager(this.mapViewer.markers, map.data.liveDataRoot + "/live/markers.json", this.events, true); return this.markerFileManager.update() .then(() => { this.markerFileManager.setAutoUpdateInterval(1000 * 10); diff --git a/common/webapp/src/js/map/LowresTileLoader.js b/common/webapp/src/js/map/LowresTileLoader.js index 632d6052c..3388683de 100644 --- a/common/webapp/src/js/map/LowresTileLoader.js +++ b/common/webapp/src/js/map/LowresTileLoader.js @@ -62,11 +62,14 @@ export class LowresTileLoader { this.geometry.translate(tileSettings.tileSize.x / 2 + 1, 0, tileSettings.tileSize.x / 2 + 1); } - load = (tileX, tileZ, cancelCheck = () => false) => { + load = (tileX, tileZ, cancelCheck = () => false, force = false) => { let tileUrl = this.tilePath + this.lod + "/" + pathFromCoords(tileX, tileZ) + '.png'; //await this.loadBlocker(); return new Promise((resolve, reject) => { + if (force) { + this.revalidatedUrls.delete(tileUrl); + } this.textureLoader.setRevalidatedUrls(this.revalidatedUrls); this.textureLoader.load(tileUrl, async texture => { diff --git a/common/webapp/src/js/map/Tile.js b/common/webapp/src/js/map/Tile.js index 375bc56c7..cf0b8e8cb 100644 --- a/common/webapp/src/js/map/Tile.js +++ b/common/webapp/src/js/map/Tile.js @@ -50,14 +50,14 @@ export class Tile { * @param tileLoader {TileLoader} * @returns {Promise} */ - load(tileLoader) { + load(tileLoader, force = false) { if (this.loading) return Promise.reject("tile is already loading!"); this.loading = true; this.unload(); this.unloaded = false; - return tileLoader.load(this.x, this.z, () => this.unloaded) + return tileLoader.load(this.x, this.z, () => this.unloaded, force) .then(model => { if (this.unloaded){ Tile.disposeModel(model); diff --git a/common/webapp/src/js/map/TileLoader.js b/common/webapp/src/js/map/TileLoader.js index 7fedf0047..49e580160 100644 --- a/common/webapp/src/js/map/TileLoader.js +++ b/common/webapp/src/js/map/TileLoader.js @@ -61,13 +61,16 @@ export class TileLoader { this.bufferGeometryLoader = new PRBMLoader(); } - load = (tileX, tileZ, cancelCheck = () => false) => { + load = (tileX, tileZ, cancelCheck = () => false, force = false) => { let tileUrl = this.tilePath + pathFromCoords(tileX, tileZ) + '.prbm'; if (this.clientDecompression) { tileUrl += '.gz'; } return new Promise((resolve, reject) => { + if (force) { + this.revalidatedUrls.delete(tileUrl); + } this.fileLoader.setRevalidatedUrls(this.revalidatedUrls); this.fileLoader.load(tileUrl, async data => { diff --git a/common/webapp/src/js/markers/MarkerManager.js b/common/webapp/src/js/markers/MarkerManager.js index 33b31cab1..500cffa90 100644 --- a/common/webapp/src/js/markers/MarkerManager.js +++ b/common/webapp/src/js/markers/MarkerManager.js @@ -37,7 +37,7 @@ export class MarkerManager { * @param fileUrl {string} - The marker file from which this manager updates its markers * @param events {EventTarget} */ - constructor(root, fileUrl, events = null) { + constructor(root, fileUrl, events = null, paused = false) { Object.defineProperty(this, 'isMarkerManager', {value: true}); this.root = root; @@ -47,6 +47,8 @@ export class MarkerManager { /** @type {NodeJS.Timeout} */ this._updateInterval = null; + this._updateIntervalMillis = 0; + this._paused = paused; } /** @@ -55,28 +57,47 @@ export class MarkerManager { * @param ms - interval in milliseconds */ setAutoUpdateInterval(ms) { + this._updateIntervalMillis = ms; if (this._updateInterval) clearTimeout(this._updateInterval); - if (ms > 0) { + if (!this._paused && ms > 0) { let autoUpdate = () => { if (this.disposed) return; this.update() .then(success => { - if (success) { - this._updateInterval = setTimeout(autoUpdate, ms); - } else { - this._updateInterval = setTimeout(autoUpdate, Math.max(ms, 1000 * 15)); + if (!this._paused){ + if (success) { + this._updateInterval = setTimeout(autoUpdate, ms); + } else { + this._updateInterval = setTimeout(autoUpdate, Math.max(ms, 1000 * 15)); + } } }) .catch(e => { alert(this.events, e, "warning"); - this._updateInterval = setTimeout(autoUpdate, Math.max(ms, 1000 * 15)); + if (!this._paused) this._updateInterval = setTimeout(autoUpdate, Math.max(ms, 1000 * 15)); }); }; - this._updateInterval = setTimeout(autoUpdate, ms); + if (!this._paused) this._updateInterval = setTimeout(autoUpdate, ms); } } + /** + * Pause auto-updates + */ + pauseAutoUpdates() { + this._paused = true; + if (this._updateInterval) clearTimeout(this._updateInterval); + } + + /** + * Resume auto-updates + */ + resumeAutoUpdates(){ + this._paused = false; + this.setAutoUpdateInterval(this._updateIntervalMillis) + } + /** * Loads the marker-file and updates all managed markers. * @returns {Promise} - A promise completing when the markers finished updating @@ -88,7 +109,6 @@ export class MarkerManager { } /** - * @protected * @param markerData */ updateFromData(markerData) {} diff --git a/common/webapp/src/js/markers/NormalMarkerManager.js b/common/webapp/src/js/markers/NormalMarkerManager.js index a9d821671..c0cda1a4b 100644 --- a/common/webapp/src/js/markers/NormalMarkerManager.js +++ b/common/webapp/src/js/markers/NormalMarkerManager.js @@ -33,8 +33,8 @@ export class NormalMarkerManager extends MarkerManager { * @param fileUrl {string} - The marker file from which this manager updates its markers * @param events {EventTarget} */ - constructor(root, fileUrl, events = null) { - super(root, fileUrl, events); + constructor(root, fileUrl, events = null, paused = false) { + super(root, fileUrl, events, paused); } /** diff --git a/common/webapp/src/js/markers/PlayerMarkerManager.js b/common/webapp/src/js/markers/PlayerMarkerManager.js index 93a99d3c2..0501c88c5 100644 --- a/common/webapp/src/js/markers/PlayerMarkerManager.js +++ b/common/webapp/src/js/markers/PlayerMarkerManager.js @@ -36,8 +36,8 @@ export class PlayerMarkerManager extends MarkerManager { * @param playerheadsUrl {string} - The url from which playerhead images should be loaded * @param events {EventTarget} */ - constructor(root, fileUrl, playerheadsUrl, events = null) { - super(root, fileUrl, events); + constructor(root, fileUrl, playerheadsUrl, events = null, paused = false) { + super(root, fileUrl, events, paused); this.playerheadsUrl = playerheadsUrl; } diff --git a/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java b/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java index dcd6085a2..c8c011dcd 100644 --- a/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java +++ b/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java @@ -40,12 +40,15 @@ import java.io.OutputStream; import java.util.Collection; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; public class HiresModelManager { private final World world; private final GridStorage storage; private final ThreadLocal> renderPasses; + private final List> tileUpdateListeners = new CopyOnWriteArrayList<>(); @Getter private final Grid tileGrid; @@ -113,6 +116,13 @@ public void unrender(Vector2i tile, TileMetaConsumer tileMetaConsumer) { ); } + public void addTileUpdateListener(Consumer listener) { + tileUpdateListeners.add(listener); + } + public void removeTileUpdateListener(Consumer listener) { + tileUpdateListeners.remove(listener); + } + private void save(final ArrayTileModel model, Vector2i tile) { try ( OutputStream out = storage.write(tile.getX(), tile.getY()); @@ -121,6 +131,12 @@ private void save(final ArrayTileModel model, Vector2i tile) { modelWriter.write(model); } catch (IOException e){ Logger.global.logError("Failed to save hires model: " + tile, e); + return; + } + + // notify listeners that the tile changed + for (Consumer listener : this.tileUpdateListeners) { + listener.accept(tile); } } diff --git a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java index a7b90b990..4473b61c0 100644 --- a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java +++ b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java @@ -39,9 +39,12 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; public class LowresLayer { @@ -61,6 +64,7 @@ public class LowresLayer { @Nullable private final LowresLayer nextLayer; private final Map pendingChanges; + private final List> tileUpdateListeners = new CopyOnWriteArrayList<>(); public LowresLayer( GridStorage storage, Grid tileGrid, int lodFactor, @@ -89,6 +93,13 @@ public LowresLayer( this.pendingChanges = new ConcurrentHashMap<>(); } + public void addTileUpdateListener(BiConsumer listener) { + tileUpdateListeners.add(listener); + } + public void removeTileUpdateListener(BiConsumer listener) { + tileUpdateListeners.remove(listener); + } + public void save() { pendingChanges.entrySet().removeIf(entry -> saveTile(entry.getKey(), entry.getValue())); if (pendingChanges.size() >= DISCARD_THRESHOLD) { @@ -130,6 +141,11 @@ private boolean saveTile(Vector2i tilePos, LowresTile tile) { return false; } + // notify listeners that the tile changed + for (BiConsumer listener : this.tileUpdateListeners) { + listener.accept(tilePos, lod); + } + if (this.nextLayer == null) return true; // write to next LOD (prepare for the most confusing grid-math you will ever see) diff --git a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java index 6e16b5457..3aeaeb225 100644 --- a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java +++ b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java @@ -24,11 +24,14 @@ */ package de.bluecolored.bluemap.core.map.lowres; +import com.flowpowered.math.vector.Vector2i; import de.bluecolored.bluemap.core.map.TileMetaConsumer; import de.bluecolored.bluemap.core.storage.MapStorage; import de.bluecolored.bluemap.core.util.Grid; import de.bluecolored.bluemap.core.util.math.Color; +import java.util.function.BiConsumer; + public class LowresTileManager implements TileMetaConsumer { private final Grid tileGrid; @@ -48,6 +51,17 @@ public LowresTileManager(MapStorage storage, Grid tileGrid, int lodCount, int lo } } + public void addTileUpdateListener(BiConsumer listener) { + for (LowresLayer layer : this.layers) { + layer.addTileUpdateListener(listener); + } + } + public void removeTileUpdateListener(BiConsumer listener) { + for (LowresLayer layer : this.layers) { + layer.removeTileUpdateListener(listener); + } + } + public synchronized void save() { for (LowresLayer layer : this.layers) { layer.save(); From b7cb2d8dd9fcd1b17b1dd71c2368c073d4097956 Mon Sep 17 00:00:00 2001 From: Lukas Rieger Date: Sat, 18 Jul 2026 14:19:23 +0200 Subject: [PATCH 02/12] Fix flickering when updating a tile --- common/webapp/src/js/map/Tile.js | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/common/webapp/src/js/map/Tile.js b/common/webapp/src/js/map/Tile.js index cf0b8e8cb..a85599dd5 100644 --- a/common/webapp/src/js/map/Tile.js +++ b/common/webapp/src/js/map/Tile.js @@ -54,8 +54,6 @@ export class Tile { if (this.loading) return Promise.reject("tile is already loading!"); this.loading = true; - this.unload(); - this.unloaded = false; return tileLoader.load(this.x, this.z, () => this.unloaded, force) .then(model => { @@ -64,6 +62,9 @@ export class Tile { return; } + this.unload(); + this.unloaded = false; + this.model = model; this.onLoad(this); }, () => { From 44618c5d9d67d2764cf18657b92ad85e8adea1a2 Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sat, 18 Jul 2026 21:31:50 -0400 Subject: [PATCH 03/12] Add header to attempt to disable buffering in upstream proxies This is an Nginx-specific header, but may be supported by other reverse proxies as well. At the very lest, it can't hurt? --- .../de/bluecolored/bluemap/common/web/MapRequestHandler.java | 4 ++++ 1 file changed, 4 insertions(+) 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 42a085a04..7ddbf2392 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 @@ -69,6 +69,10 @@ public MapRequestHandler( HttpResponse response = new HttpResponse(HttpStatusCode.OK); response.addHeader("Content-Type", "text/event-stream"); response.addHeader("Cache-Control", "no-cache"); + + // attempt to turn off buffering in upstream proxy + response.addHeader("X-Accel-Buffering", "no"); + try { response.setBody(sseConnections.openConnection()); } catch (IOException e) { From a1282622407f50d1f982dfa8dfcc46c8027ea21c Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sat, 18 Jul 2026 22:11:27 -0400 Subject: [PATCH 04/12] Move LiveDataSupplierBroadcaster's polling to the BlueMap.THREAD_POOL The polling of data at a fixed interval is now scheduled by the BlueMap.SCHEDULER and the work of doing the updates is run in the BlueMap.THREAD_POOL. This makes the LiveDataSupplierBroadcaster only responsible for starting/stopping the scheduled task when listeners connect/disconnect. This avoids the overhead of having to spin up a polling thread for every supplier * every map, as well is a much simpler implementation. --- .../web/LiveDataSupplierBroadcaster.java | 66 +++++++++---------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java b/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java index 3810239fe..85b5a14b3 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java @@ -24,66 +24,62 @@ */ package de.bluecolored.bluemap.common.web; +import de.bluecolored.bluemap.core.BlueMap; + import java.io.Closeable; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.Supplier; /** - * Polls a {@link Supplier} on a background thread and notifies registered listeners - * whenever the returned value changes. + * Polls a {@link Supplier} and notifies registered listeners whenever the + * returned value changes. + * + * Polling and updating the data is done by the {@link BlueMap#SCHEDULER} and + * {@link BlueMap#THREAD_POOL}. * * Call {@link #update()} to get the current value (rate-limited by the polling rate) - * Call {@link #close()} to stop the background thread. + * Call {@link #close()} to stop the background polling. */ public class LiveDataSupplierBroadcaster implements Supplier, Closeable { private final Supplier dataSupplier; private final long pollIntervalMillis; private final Set> listeners = ConcurrentHashMap.newKeySet(); - private final Thread pollThread; - private volatile boolean closed = false; + private ScheduledFuture pollTask = null; + private boolean closed = false; private long lastUpdate = -1; private T data = null; public LiveDataSupplierBroadcaster(Supplier dataSupplier, long pollIntervalMillis) { this.dataSupplier = dataSupplier; this.pollIntervalMillis = pollIntervalMillis; - this.pollThread = new Thread(this::pollLoop, String.format("%sPoller", dataSupplier.getClass().getName())); - this.pollThread.setDaemon(true); - this.pollThread.start(); } public synchronized void addUpdateListener(Consumer listener) { listeners.add(listener); - notifyAll(); - } - public void removeUpdateListener(Consumer listener) { - listeners.remove(listener); + // have a listener - ensure scheduled poll task is running + if (!closed && pollTask == null) { + pollTask = BlueMap.SCHEDULER.scheduleWithFixedDelay( + () -> BlueMap.THREAD_POOL.execute(this::update), + 0, + pollIntervalMillis, + TimeUnit.MILLISECONDS + ); + } } - private void pollLoop() { - while (!closed) { - // suspend polling until at least one listener is registered - synchronized (this) { - try { - while (listeners.isEmpty() && !closed){ - wait(); - } - } catch (InterruptedException ignored){ - break; - } - } - if (closed) break; - update(); + public synchronized void removeUpdateListener(Consumer listener) { + listeners.remove(listener); - try { - Thread.sleep(pollIntervalMillis); - } catch (InterruptedException ignored) { - break; - } + // stop automatically updating if nothing is listening anymore + if (listeners.isEmpty() && pollTask != null) { + pollTask.cancel(false); + pollTask = null; } } @@ -117,13 +113,15 @@ private void update() { } /** - * Stops the background polling thread. + * Stops the background polling. */ @Override public synchronized void close() { closed = true; - notifyAll(); - pollThread.interrupt(); + if (pollTask != null) { + pollTask.cancel(false); + pollTask = null; + } } } From 6253d809cccf8f6def1e00d79d3875b1730b4f29 Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sun, 19 Jul 2026 02:16:34 -0400 Subject: [PATCH 05/12] Refactor SSE connections to improve handling of slow consumers Previously a single slow consumer of events would block all other consumers. This was because the `SseConnectionManager` would handle sending data to each connection in a single worker thread. This commit refactors the code so that the `SseConnectionManager` is no longer responsible for actually sending the data to each connection. Instead, it simply enqueues events to every managed `SseConnection` which handles sending them. Details: - Each `SseConnection` now has a queue and a virtual thread that pulls events off it and sends them to the consumer. This means that a slow consumer now only blocks its own connection. - The per-connection queues now limit how many events can be queued to send. After the limit is reached, any new events are simply dropped to avoid slow clients forcing the server to buffer an unlimited number of old events. - An `onClose` callback can now be registered on the `SseConnection` class. This allows it to own its entire lifecycle, while still allowing the `SseConnectionManager` to be notified when it's been closed and needs removing from the set of managed connections. --- .../bluemap/common/web/SseConnection.java | 74 +++++++++++++-- .../common/web/SseConnectionManager.java | 89 +++++-------------- 2 files changed, 89 insertions(+), 74 deletions(-) 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 d8e08e5dc..efdd58aaa 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 @@ -25,11 +25,14 @@ package de.bluecolored.bluemap.common.web; import java.io.Closeable; +import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; import java.io.PipedInputStream; import java.io.PipedOutputStream; import java.nio.charset.StandardCharsets; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.BlockingQueue; import lombok.SneakyThrows; @@ -37,22 +40,47 @@ * 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 sent to it via {@link #send(String, String)}. - * Sending an event will flush the stream, ensuring that all events can be read immediately. + * 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. */ public class SseConnection implements Closeable { - private static final int PIPE_BUFFER_SIZE = 4096; + private static final int PIPE_BUFFER_SIZE = 1024; + + // 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 PipedInputStream pipeIn; + 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 PipedInputStream(pipeOut, PIPE_BUFFER_SIZE); + this.pipeIn = new FilterInputStream(new PipedInputStream(pipeOut, PIPE_BUFFER_SIZE)) { + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + SseConnection.this.close(); + } + } + }; + + 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; } @@ -61,6 +89,38 @@ public boolean isClosed() { return closed; } + /** + * Registers a callback that's called when this connection closes. + * + * Returns true if the regsitration suceeded, false if the connection was already closed. + */ + public synchronized boolean setOnClose(Runnable onClose) { + if (closed) return false; + this.onClose = onClose; + return true; + } + + /** + * Queues an SSE event to be delivered to this connection. + * + * If this connection's queue is full (due to a slowly-reading client), the event is + * silently dropped. + */ + public void enqueue(String eventType, String data) { + if (closed) return; + // TODO: close the connection if the queue is full to force a reconnect? + queue.offer(new String[]{eventType, data}); + } + + private void sendLoop() { + try { + while (!closed) { + String[] event = queue.take(); + send(event[0], event[1]); + } + } catch (InterruptedException | IOException ignored) {} + } + @SneakyThrows(IOException.class) // allows using this function in the forEach below private void writeLine(String line){ pipeOut.write((line + "\n").getBytes(StandardCharsets.UTF_8)); @@ -71,7 +131,7 @@ private void writeLine(String line){ * * @throws IOException if the connection is closed or the client has disconnected */ - public synchronized void send(String eventType, String data) throws IOException { + private synchronized void send(String eventType, String data) throws IOException { if (closed) throw new IOException("SSE connection is closed"); try { writeLine("event: " + eventType); @@ -88,7 +148,9 @@ public synchronized void send(String eventType, String data) throws IOException public synchronized void close() { if (closed) return; closed = true; + sendThread.interrupt(); try { pipeOut.close(); } catch (IOException ignored) {} + 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 cf8d27920..dd03c475a 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 @@ -25,24 +25,21 @@ package de.bluecolored.bluemap.common.web; import java.io.Closeable; -import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; -import java.util.ArrayList; -import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.LinkedBlockingQueue; import java.util.function.Consumer; /** - * Tracks active {@link SseConnection}s and provides queued broadcast delivery. + * Tracks active {@link SseConnection}s and provides broadcast delivery. + * + * Broadcasting just queues events into each managed {@link SseConnection} without + * blocking so a slow or stuck client only affects itself. */ public class SseConnectionManager implements Closeable { private final Set connections = ConcurrentHashMap.newKeySet(); - private final LinkedBlockingQueue broadcastQueue = new LinkedBlockingQueue<>(); - private final Thread broadcastThread; private volatile boolean closed; // allow objects to listen for when the connection count transitions between 0 and non-0 @@ -51,12 +48,6 @@ public class SseConnectionManager implements Closeable { public void addHasConnectionsListener(Consumer listener) { hasConnectionsListeners.add(listener); } public void removeHasConnectionsListener(Consumer listener) { hasConnectionsListeners.remove(listener); } - public SseConnectionManager() { - this.broadcastThread = new Thread(this::broadcastLoop, "bluemap-sse-broadcast"); - this.broadcastThread.setDaemon(true); - this.broadcastThread.start(); - } - /** * 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 @@ -66,24 +57,23 @@ public SseConnectionManager() { public InputStream openConnection() throws IOException { SseConnection connection = new SseConnection(); add(connection); - return new FilterInputStream(connection.getInputStream()) { - @Override - public void close() throws IOException { - try { - super.close(); - } finally { - connection.close(); - remove(connection); - } - } - }; + return connection.getInputStream(); } public void add(SseConnection connection) { boolean fire; synchronized (this) { fire = connections.isEmpty(); - fire = connections.add(connection) && fire; + if (connections.add(connection)){ + if (!connection.setOnClose(() -> remove(connection))){ + // failed to register an onClose callback (connection is already closed?) + connections.remove(connection); + fire = false; + } + } + else { + fire = false; + } } if (fire) notifyHasConnections(true); } @@ -97,7 +87,7 @@ public void remove(SseConnection connection) { } /** - * Queues an SSE event to be sent to all live connections on the background broadcast thread. + * Broadcast an SSE event to all live connections. * Returns immediately without blocking. * * @param eventType the SSE event type @@ -105,58 +95,21 @@ public void remove(SseConnection connection) { */ public void broadcast(String eventType, String data) { if (closed) return; - broadcastQueue.offer(new String[]{eventType, data}); - } - - private void broadcastLoop() { - try { - while (!closed) { - String[] event = broadcastQueue.take(); - broadcastSync(event[0], event[1]); - } - } catch (InterruptedException ignored) {} - } - - /** - * Sends an SSE event to all live connections synchronously. - * Dead or broken connections are removed automatically. - */ - private void broadcastSync(String eventType, String data) { - if (connections.isEmpty()) return; - List toRemove = new ArrayList<>(); for (SseConnection conn : connections) { - try { - conn.send(eventType, data); - } catch (IOException ignored) {} - if (conn.isClosed()) { - toRemove.add(conn); - } - } - if (!toRemove.isEmpty()) { - boolean fire; - synchronized (this) { - fire = connections.removeAll(toRemove) && connections.isEmpty(); - } - if (fire) notifyHasConnections(false); + conn.enqueue(eventType, data); } } /** - * Closes all registered connections, clears the registry, and stops the broadcast thread. + * Closes all registered connections. Each connection removes itself from the registry + * (via {@link #remove(SseConnection)}) as it closes. */ @Override public void close() { closed = true; - broadcastThread.interrupt(); - boolean fire; - synchronized (this) { - fire = !connections.isEmpty(); - for (SseConnection conn : connections) { - conn.close(); - } - connections.clear(); + for (SseConnection conn : connections) { + conn.close(); } - if (fire) notifyHasConnections(false); } private void notifyHasConnections(boolean hasConnections) { From d2b70951f4b3712489792bedf4a613b86ae54858 Mon Sep 17 00:00:00 2001 From: Lukas Rieger Date: Sun, 19 Jul 2026 11:22:16 +0200 Subject: [PATCH 06/12] Use OnCloseInputStream and normalize thread-name --- .../bluemap/common/web/SseConnection.java | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) 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 efdd58aaa..d8b93f24e 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 @@ -25,7 +25,6 @@ package de.bluecolored.bluemap.common.web; import java.io.Closeable; -import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; import java.io.PipedInputStream; @@ -34,6 +33,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.BlockingQueue; +import de.bluecolored.bluemap.core.util.stream.OnCloseInputStream; import lombok.SneakyThrows; /** @@ -63,18 +63,9 @@ public class SseConnection implements Closeable { 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 FilterInputStream(new PipedInputStream(pipeOut, PIPE_BUFFER_SIZE)) { - @Override - public void close() throws IOException { - try { - super.close(); - } finally { - SseConnection.this.close(); - } - } - }; + this.pipeIn = new OnCloseInputStream(new PipedInputStream(pipeOut, PIPE_BUFFER_SIZE), SseConnection.this); - this.sendThread = Thread.ofVirtual().name("bluemap-sse-send").start(this::sendLoop); + this.sendThread = Thread.ofVirtual().name("BlueMap-SSE-send").start(this::sendLoop); } /** From 0a851f9dba7cbf5d51bb76baf3a8f946fa787baf Mon Sep 17 00:00:00 2001 From: Lukas Rieger Date: Sun, 19 Jul 2026 12:15:15 +0200 Subject: [PATCH 07/12] Fix small IDE warnings and formatting nitpicks --- .../common/web/LiveDataSupplierBroadcaster.java | 6 +++--- .../bluemap/common/web/MapRequestHandler.java | 7 +++---- .../bluemap/common/web/SseConnection.java | 15 +++++++-------- .../bluemap/common/web/SseConnectionManager.java | 16 +++++++++------- .../core/map/hires/HiresModelManager.java | 1 + .../bluemap/core/map/lowres/LowresLayer.java | 1 + .../core/map/lowres/LowresTileManager.java | 1 + 7 files changed, 25 insertions(+), 22 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java b/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java index 85b5a14b3..3e513a44c 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/LiveDataSupplierBroadcaster.java @@ -37,10 +37,10 @@ /** * Polls a {@link Supplier} and notifies registered listeners whenever the * returned value changes. - * + *

* Polling and updating the data is done by the {@link BlueMap#SCHEDULER} and * {@link BlueMap#THREAD_POOL}. - * + *

* Call {@link #update()} to get the current value (rate-limited by the polling rate) * Call {@link #close()} to stop the background polling. */ @@ -85,7 +85,7 @@ public synchronized void removeUpdateListener(Consumer listener) { /** * Ensure the data is up to date and return it. - * + *

* Note that this will only get new data from the supplier if the current * cached data is stale (according to {@code pollIntervalMillis}). */ 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 7ddbf2392..f9462874d 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 @@ -24,7 +24,6 @@ */ package de.bluecolored.bluemap.common.web; -import de.bluecolored.bluemap.common.web.http.HttpRequestHandler; import de.bluecolored.bluemap.common.web.http.HttpResponse; import de.bluecolored.bluemap.common.web.http.HttpStatusCode; import de.bluecolored.bluemap.core.map.BmMap; @@ -49,9 +48,9 @@ public MapRequestHandler( this(map.getStorage(), livePlayersDataSupplier, liveMarkerDataSupplier); // only register the handler for map updates if we're given the actual map - // instance from the plugin (ie. not running standalone) + // instance from the plugin (i.e. not running standalone) map.getHiresModelManager().addTileUpdateListener(tile -> onTileUpdate(tile, 0)); - map.getLowresTileManager().addTileUpdateListener((tile, lod) -> onTileUpdate(tile, lod)); + map.getLowresTileManager().addTileUpdateListener(this::onTileUpdate); } public MapRequestHandler(MapStorage mapStorage) { @@ -65,7 +64,7 @@ public MapRequestHandler( ) { register(".*", new MapStorageRequestHandler(mapStorage)); - register("live/sse", "", (HttpRequestHandler) request -> { + register("live/sse", "", _ -> { HttpResponse response = new HttpResponse(HttpStatusCode.OK); response.addHeader("Content-Type", "text/event-stream"); response.addHeader("Cache-Control", "no-cache"); 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 d8b93f24e..47dfdc360 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 @@ -38,10 +38,10 @@ /** * 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. */ @@ -82,7 +82,7 @@ public boolean isClosed() { /** * Registers a callback that's called when this connection closes. - * + *

* Returns true if the regsitration suceeded, false if the connection was already closed. */ public synchronized boolean setOnClose(Runnable onClose) { @@ -93,14 +93,13 @@ public synchronized boolean setOnClose(Runnable onClose) { /** * Queues an SSE event to be delivered to this connection. - * + *

* If this connection's queue is full (due to a slowly-reading client), the event is * silently dropped. */ - public void enqueue(String eventType, String data) { - if (closed) return; - // TODO: close the connection if the queue is full to force a reconnect? - queue.offer(new String[]{eventType, data}); + public boolean enqueue(String eventType, String data) { + if (closed) return false; + return queue.offer(new String[]{eventType, data}); } private void sendLoop() { 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 dd03c475a..4f462adb5 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 @@ -33,7 +33,7 @@ /** * Tracks active {@link SseConnection}s and provides broadcast delivery. - * + *

* Broadcasting just queues events into each managed {@link SseConnection} without * blocking so a slow or stuck client only affects itself. */ @@ -45,8 +45,13 @@ public class SseConnectionManager implements Closeable { // allow objects to listen for when the connection count transitions between 0 and non-0 private final Set> hasConnectionsListeners = ConcurrentHashMap.newKeySet(); - public void addHasConnectionsListener(Consumer listener) { hasConnectionsListeners.add(listener); } - public void removeHasConnectionsListener(Consumer listener) { hasConnectionsListeners.remove(listener); } + public void addHasConnectionsListener(Consumer listener) { + hasConnectionsListeners.add(listener); + } + + public void removeHasConnectionsListener(Consumer listener) { + hasConnectionsListeners.remove(listener); + } /** * Creates a new {@link SseConnection}, registers it, and returns an {@link InputStream} suitable @@ -64,16 +69,13 @@ public void add(SseConnection connection) { boolean fire; synchronized (this) { fire = connections.isEmpty(); - if (connections.add(connection)){ + if (connections.add(connection)) { if (!connection.setOnClose(() -> remove(connection))){ // failed to register an onClose callback (connection is already closed?) connections.remove(connection); fire = false; } } - else { - fire = false; - } } if (fire) notifyHasConnections(true); } diff --git a/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java b/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java index c8c011dcd..2cfb8d281 100644 --- a/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java +++ b/core/src/main/java/de/bluecolored/bluemap/core/map/hires/HiresModelManager.java @@ -119,6 +119,7 @@ public void unrender(Vector2i tile, TileMetaConsumer tileMetaConsumer) { public void addTileUpdateListener(Consumer listener) { tileUpdateListeners.add(listener); } + public void removeTileUpdateListener(Consumer listener) { tileUpdateListeners.remove(listener); } diff --git a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java index 4473b61c0..f782a155a 100644 --- a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java +++ b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresLayer.java @@ -96,6 +96,7 @@ public LowresLayer( public void addTileUpdateListener(BiConsumer listener) { tileUpdateListeners.add(listener); } + public void removeTileUpdateListener(BiConsumer listener) { tileUpdateListeners.remove(listener); } diff --git a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java index 3aeaeb225..e99b19023 100644 --- a/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java +++ b/core/src/main/java/de/bluecolored/bluemap/core/map/lowres/LowresTileManager.java @@ -56,6 +56,7 @@ public void addTileUpdateListener(BiConsumer listener) { layer.addTileUpdateListener(listener); } } + public void removeTileUpdateListener(BiConsumer listener) { for (LowresLayer layer : this.layers) { layer.removeTileUpdateListener(listener); From c0b5cadd6d647f30dbf8f202f22124242766df65 Mon Sep 17 00:00:00 2001 From: Lukas Rieger Date: Sun, 19 Jul 2026 16:51:09 +0200 Subject: [PATCH 08/12] Close SSE connection if event-queue overflows --- .../bluecolored/bluemap/common/web/SseConnection.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) 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 47dfdc360..e211cb795 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 @@ -95,11 +95,13 @@ public synchronized boolean setOnClose(Runnable onClose) { * Queues an SSE event to be delivered to this connection. *

* If this connection's queue is full (due to a slowly-reading client), the event is - * silently dropped. + * silently dropped and the connection will be closed. */ - public boolean enqueue(String eventType, String data) { - if (closed) return false; - return queue.offer(new String[]{eventType, data}); + public void enqueue(String eventType, String data) { + if (closed) return; + if (!queue.offer(new String[]{eventType, data})) { + close(); + } } private void sendLoop() { From 2fe878755cd1f9299f93415196d60d0eda7d03da Mon Sep 17 00:00:00 2001 From: Lukas Rieger Date: Sun, 19 Jul 2026 22:14:00 +0200 Subject: [PATCH 09/12] Add SSE config option, and add sse-support also for CLI --- .../common/config/WebserverConfig.java | 37 ++----------- .../bluemap/common/plugin/Plugin.java | 2 +- .../bluemap/common/web/MapRequestHandler.java | 55 ++++++++++--------- .../bluecolored/bluemap/config/webserver.conf | 4 ++ .../bluecolored/bluemap/cli/BlueMapCLI.java | 39 ++++++++----- 5 files changed, 64 insertions(+), 73 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/config/WebserverConfig.java b/common/src/main/java/de/bluecolored/bluemap/common/config/WebserverConfig.java index 533f69514..4cef22e5b 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/config/WebserverConfig.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/config/WebserverConfig.java @@ -24,6 +24,7 @@ */ package de.bluecolored.bluemap.common.config; +import lombok.Getter; import org.spongepowered.configurate.objectmapping.ConfigSerializable; import java.net.InetAddress; @@ -33,6 +34,7 @@ @SuppressWarnings({"FieldMayBeFinal", "FieldCanBeLocal"}) @ConfigSerializable +@Getter public class WebserverConfig { private boolean enabled = true; @@ -41,19 +43,9 @@ public class WebserverConfig { private String ip = "0.0.0.0"; private int port = 8100; - private LogConfig log = new LogConfig(); - - public boolean isEnabled() { - return enabled; - } - - public Path getWebroot() { - return webroot; - } + private boolean sseEnabled = true; - public String getIp() { - return ip; - } + private LogConfig log = new LogConfig(); public InetAddress resolveIp() throws UnknownHostException { if (ip.isEmpty() || ip.equals("0.0.0.0") || ip.equals("::0")) { @@ -65,33 +57,14 @@ public InetAddress resolveIp() throws UnknownHostException { } } - public int getPort() { - return port; - } - - public LogConfig getLog() { - return log; - } - @ConfigSerializable + @Getter public static class LogConfig { private String file = null; private boolean append = false; private String format = "%1$s \"%3$s %4$s %5$s\" %6$s %7$s"; - public String getFile() { - return file; - } - - public boolean isAppend() { - return append; - } - - public String getFormat() { - return format; - } - } } diff --git a/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java b/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java index 876808001..c9596860e 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/plugin/Plugin.java @@ -234,7 +234,7 @@ private void load(@Nullable ResourcePack preloadedResourcePack) throws IOExcepti null; LiveMarkersDataSupplier liveMarkersDataSupplier = new LiveMarkersDataSupplier(map.getMarkerSets()); - mapRequestHandler = new MapRequestHandler(map, livePlayersDataSupplier, liveMarkersDataSupplier); + mapRequestHandler = new MapRequestHandler(map, livePlayersDataSupplier, liveMarkersDataSupplier, webserverConfig.isSseEnabled()); } else { Storage storage = blueMap.getOrLoadStorage(mapConfig.getStorage()); mapRequestHandler = new MapRequestHandler(storage.map(id)); 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 f9462874d..f0e4603c1 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 @@ -43,52 +43,56 @@ public class MapRequestHandler extends RoutingRequestHandler { public MapRequestHandler( BmMap map, @Nullable Supplier livePlayersDataSupplier, - @Nullable Supplier liveMarkerDataSupplier + @Nullable Supplier liveMarkerDataSupplier, + boolean useSSE ) { - this(map.getStorage(), livePlayersDataSupplier, liveMarkerDataSupplier); + this(map.getStorage(), livePlayersDataSupplier, liveMarkerDataSupplier, useSSE); - // only register the handler for map updates if we're given the actual map - // instance from the plugin (i.e. not running standalone) - map.getHiresModelManager().addTileUpdateListener(tile -> onTileUpdate(tile, 0)); - map.getLowresTileManager().addTileUpdateListener(this::onTileUpdate); + if (useSSE) { + map.getHiresModelManager().addTileUpdateListener(tile -> onTileUpdate(tile, 0)); + map.getLowresTileManager().addTileUpdateListener(this::onTileUpdate); + } } public MapRequestHandler(MapStorage mapStorage) { - this(mapStorage, null, null); + this(mapStorage, null, null, false); } public MapRequestHandler( MapStorage mapStorage, @Nullable Supplier livePlayersDataSupplier, - @Nullable Supplier liveMarkerDataSupplier + @Nullable Supplier liveMarkerDataSupplier, + boolean useSSE ) { register(".*", new MapStorageRequestHandler(mapStorage)); - register("live/sse", "", _ -> { - HttpResponse response = new HttpResponse(HttpStatusCode.OK); - response.addHeader("Content-Type", "text/event-stream"); - response.addHeader("Cache-Control", "no-cache"); - - // 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); - } - return response; - }); + if (useSSE) { + register("live/sse", "", _ -> { + HttpResponse response = new HttpResponse(HttpStatusCode.OK); + response.addHeader("Content-Type", "text/event-stream"); + response.addHeader("Cache-Control", "no-cache"); + + // 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); + } + return response; + }); + } if (livePlayersDataSupplier != null) { LiveDataSupplierBroadcaster playerDataBroadcaster = new LiveDataSupplierBroadcaster<>(livePlayersDataSupplier, 1000); - registerSseCallback(playerDataBroadcaster, this::onPlayerUpdate); + if (useSSE) registerSseCallback(playerDataBroadcaster, this::onPlayerUpdate); register("live/players\\.json", "", new JsonDataRequestHandler(playerDataBroadcaster)); } if (liveMarkerDataSupplier != null) { LiveDataSupplierBroadcastermarkerDataBroadcaster = new LiveDataSupplierBroadcaster<>(liveMarkerDataSupplier, 10000); - registerSseCallback(markerDataBroadcaster, this::onMarkerUpdate); + if (useSSE) registerSseCallback(markerDataBroadcaster, this::onMarkerUpdate); register("live/markers\\.json", "", new JsonDataRequestHandler(markerDataBroadcaster)); } } @@ -119,4 +123,5 @@ private void onPlayerUpdate(String data) { private void onMarkerUpdate(String data) { sseConnections.broadcast("marker", data); } + } diff --git a/common/src/main/resources/de/bluecolored/bluemap/config/webserver.conf b/common/src/main/resources/de/bluecolored/bluemap/config/webserver.conf index 44174b457..ad85f7fda 100644 --- a/common/src/main/resources/de/bluecolored/bluemap/config/webserver.conf +++ b/common/src/main/resources/de/bluecolored/bluemap/config/webserver.conf @@ -17,6 +17,10 @@ webroot: "${webroot}" # Default is 8100 port: 8100 +# Whether to use Server-Sent Events (SSE) for pushing tile and marker-updates to the connected clients. +# Default is true +sse-enabled: true + # Config-section for webserver activity logging: log: { # The file where all the webserver activity will be logged to. diff --git a/implementations/cli/src/main/java/de/bluecolored/bluemap/cli/BlueMapCLI.java b/implementations/cli/src/main/java/de/bluecolored/bluemap/cli/BlueMapCLI.java index 903329dab..4cd02e4b0 100644 --- a/implementations/cli/src/main/java/de/bluecolored/bluemap/cli/BlueMapCLI.java +++ b/implementations/cli/src/main/java/de/bluecolored/bluemap/cli/BlueMapCLI.java @@ -32,6 +32,7 @@ import de.bluecolored.bluemap.common.api.BlueMapAPIImpl; import de.bluecolored.bluemap.common.commands.TextFormat; import de.bluecolored.bluemap.common.config.*; +import de.bluecolored.bluemap.common.live.LiveMarkersDataSupplier; import de.bluecolored.bluemap.common.metrics.Metrics; import de.bluecolored.bluemap.common.plugin.MapUpdateService; import de.bluecolored.bluemap.common.rendermanager.MapUpdatePreparationTask; @@ -238,15 +239,20 @@ public void run() { ); // wait until done, then shutdown if not watching - renderManager.awaitIdle(); - if (shutdownInProgress) return; + Thread.ofPlatform().name("BlueMap-CLI-Render").start(() -> { + try { + renderManager.awaitIdle(); + } catch (InterruptedException ignore) {} - Logger.global.logInfo("Your maps are now all up-to-date!"); + if (shutdownInProgress) return; - if (!watch) { - Runtime.getRuntime().removeShutdownHook(shutdownHook); - shutdown.run(); - } + Logger.global.logInfo("Your maps are now all up-to-date!"); + + if (!watch) { + Runtime.getRuntime().removeShutdownHook(shutdownHook); + shutdown.run(); + } + }); } public void updateMarkers(BlueMapService blueMap, @Nullable String mapsToUpdate) { @@ -292,11 +298,16 @@ public void startWebserver(BlueMapService blueMap, boolean verbose) throws IOExc for (var mapConfigEntry : blueMap.getConfig().getMapConfigs().entrySet()) { MapStorage storage = blueMap.getOrLoadStorage(mapConfigEntry.getValue().getStorage()) .map(mapConfigEntry.getKey()); + BmMap map = blueMap.getMaps().get(mapConfigEntry.getKey()); + + MapRequestHandler mapRequestHandler = map != null ? + new MapRequestHandler(map, null, new LiveMarkersDataSupplier(map.getMarkerSets()), config.isSseEnabled()) : + new MapRequestHandler(storage); routingRequestHandler.register( "maps/" + Pattern.quote(mapConfigEntry.getKey()) + "/(.*)", "$1", - new MapRequestHandler(storage) + mapRequestHandler ); } @@ -418,13 +429,6 @@ public static void main(String[] args) { blueMap = new BlueMapService(configs); boolean noActions = true; - if (cmd.hasOption("w")) { - noActions = false; - - cli.startWebserver(blueMap, cmd.hasOption("b")); - Thread.sleep(1000); //wait a second to let the webserver start, looks nicer in the log if anything comes after that - } - if (cmd.hasOption("r") || cmd.hasOption("f") || cmd.hasOption("u") || cmd.hasOption("e")) { noActions = false; @@ -451,6 +455,11 @@ public static void main(String[] args) { } } + if (cmd.hasOption("w")) { + noActions = false; + cli.startWebserver(blueMap, cmd.hasOption("b")); + } + // if nothing has been defined to do if (noActions) { Logger.global.logInfo("Generated default config files for you, here: " + cli.configFolder.toAbsolutePath().normalize() + "\n"); From 4a4ea637a1d9c547283210b5e51739bdfc61fc6f Mon Sep 17 00:00:00 2001 From: Lukas Rieger Date: Sun, 19 Jul 2026 22:21:53 +0200 Subject: [PATCH 10/12] Fix error when pressing "Update Map" while recieving tile events --- common/webapp/src/js/BlueMapApp.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/webapp/src/js/BlueMapApp.js b/common/webapp/src/js/BlueMapApp.js index 6017dcf44..958029834 100644 --- a/common/webapp/src/js/BlueMapApp.js +++ b/common/webapp/src/js/BlueMapApp.js @@ -434,7 +434,7 @@ export class BlueMapApp { const parsed = JSON.parse(data); const mgr = parsed.lod > 0 ? map.lowresTileManager[parsed.lod - 1] : map.hiresTileManager; - if (!mgr.unloaded) { + if (mgr && !mgr.unloaded) { const tilehash = hashTile(parsed.x, parsed.y); const tile = mgr.tiles.get(tilehash); if (tile && !tile.loading) { From 0b222802b0d5e7d13d22002786ac4c0fb6f79708 Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sun, 19 Jul 2026 21:30:29 -0400 Subject: [PATCH 11/12] Fix disposed MarkerManagers being able to be unpaused --- common/webapp/src/js/markers/MarkerManager.js | 1 + 1 file changed, 1 insertion(+) diff --git a/common/webapp/src/js/markers/MarkerManager.js b/common/webapp/src/js/markers/MarkerManager.js index 500cffa90..59bb27db7 100644 --- a/common/webapp/src/js/markers/MarkerManager.js +++ b/common/webapp/src/js/markers/MarkerManager.js @@ -94,6 +94,7 @@ export class MarkerManager { * Resume auto-updates */ resumeAutoUpdates(){ + if (this.disposed) return; this._paused = false; this.setAutoUpdateInterval(this._updateIntervalMillis) } From 93e13276c893919be9880ad310923ca90f1981c2 Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Sun, 19 Jul 2026 21:52:44 -0400 Subject: [PATCH 12/12] Remove verbose logging of all received SSE --- common/webapp/src/js/BlueMapApp.js | 3 --- 1 file changed, 3 deletions(-) diff --git a/common/webapp/src/js/BlueMapApp.js b/common/webapp/src/js/BlueMapApp.js index 958029834..7d85fc3fc 100644 --- a/common/webapp/src/js/BlueMapApp.js +++ b/common/webapp/src/js/BlueMapApp.js @@ -430,7 +430,6 @@ export class BlueMapApp { }); this.mapEventSource.addEventListener("tile", ({ data }) => { - alert(this.events, `tile update: ${data}`, "debug"); const parsed = JSON.parse(data); const mgr = parsed.lod > 0 ? map.lowresTileManager[parsed.lod - 1] : map.hiresTileManager; @@ -444,12 +443,10 @@ export class BlueMapApp { }); this.mapEventSource.addEventListener("player", ({ data }) => { - alert(this.events, `player update: ${data}`, "debug"); this.playerMarkerManager.updateFromData(JSON.parse(data)); }); this.mapEventSource.addEventListener("marker", ({ data }) => { - alert(this.events, `marker update: ${data}`, "debug"); this.markerFileManager.updateFromData(JSON.parse(data)); }); }