From ea33744ec96330077acca1d451e9b85f207ea549 Mon Sep 17 00:00:00 2001 From: iceBear67 Date: Wed, 15 Jul 2026 16:43:43 +0800 Subject: [PATCH] refactor: logging and improved multiplex handling --- server/build.gradle.kts | 7 + .../java/io/icybear/redapricot/Config.java | 51 +++--- .../io/icybear/redapricot/ControlSession.java | 23 ++- .../main/java/io/icybear/redapricot/Hub.java | 35 ++-- .../io/icybear/redapricot/HubConnection.java | 36 +++-- .../java/io/icybear/redapricot/HubServer.java | 19 ++- .../main/java/io/icybear/redapricot/Main.java | 14 +- .../io/icybear/redapricot/PendingPlayer.java | 33 ++-- .../io/icybear/redapricot/WorkerConn.java | 94 +++++++---- .../redapricot/net/EncryptedFrames.java | 5 + .../java/io/icybear/redapricot/util/Json.java | 151 ------------------ .../icybear/redapricot/util/ProtoReader.java | 69 ++++---- server/src/main/resources/log4j2.xml | 16 ++ 13 files changed, 238 insertions(+), 315 deletions(-) delete mode 100644 server/src/main/java/io/icybear/redapricot/util/Json.java create mode 100644 server/src/main/resources/log4j2.xml diff --git a/server/build.gradle.kts b/server/build.gradle.kts index 5b25ca3..44458c8 100644 --- a/server/build.gradle.kts +++ b/server/build.gradle.kts @@ -16,6 +16,8 @@ repositories { dependencies { implementation("io.vertx:vertx-core:5.1.5") + implementation("org.apache.logging.log4j:log4j-core:2.26.0") + implementation("org.apache.logging.log4j:log4j-jul:2.26.0") // route Netty/JUL fallback -> log4j2 testImplementation("org.junit.jupiter:junit-jupiter:5.10.2") testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.10.2") } @@ -28,6 +30,11 @@ java { application { mainClass = "io.icybear.redapricot.Main" + // Route java.util.logging (Netty's fallback) through Log4j2 on the `run`/`installDist` + // launch paths. The `java -jar` shadow-jar path sets this programmatically in Main. + applicationDefaultJvmArgs = listOf( + "-Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager" + ) } tasks.test { diff --git a/server/src/main/java/io/icybear/redapricot/Config.java b/server/src/main/java/io/icybear/redapricot/Config.java index 08e6d77..a3d5e8a 100644 --- a/server/src/main/java/io/icybear/redapricot/Config.java +++ b/server/src/main/java/io/icybear/redapricot/Config.java @@ -1,48 +1,35 @@ package io.icybear.redapricot; -import io.icybear.redapricot.util.Json; +import io.vertx.core.json.JsonObject; import java.nio.file.Files; import java.nio.file.Path; -import java.util.Map; /** Hub configuration, loaded from a JSON file. See PROTOCOL.md §9.1. */ -public final class Config { - public final String host; - public final int port; - public final String psk; - public final long timestampWindowMs; - public final long pendingTimeoutMs; - - public Config(String host, int port, String psk, long timestampWindowMs, long pendingTimeoutMs) { - this.host = host; - this.port = port; - this.psk = psk; - this.timestampWindowMs = timestampWindowMs; - this.pendingTimeoutMs = pendingTimeoutMs; - } - +public record Config( + String host, + int port, + String psk, + long timestampWindowMs, + long pendingTimeoutMs +) { public static Config load(Path file) throws Exception { - Map m = Json.parseObject(Files.readString(file)); - String listen = str(m, "listen", "0.0.0.0:25565"); + JsonObject json = new JsonObject(Files.readString(file)); + + String listen = json.getString("listen", "0.0.0.0:25565"); int idx = listen.lastIndexOf(':'); if (idx < 0) throw new IllegalArgumentException("listen must be host:port"); String host = listen.substring(0, idx); int port = Integer.parseInt(listen.substring(idx + 1)); - String psk = str(m, "psk", null); + + String psk = json.getString("psk"); if (psk == null || psk.isEmpty()) throw new IllegalArgumentException("psk is required"); - long tsWin = num(m, "timestampWindowMs", 30000); - long pending = num(m, "pendingTimeoutMs", 10000); - return new Config(host, port, psk, tsWin, pending); - } - private static String str(Map m, String k, String def) { - Object v = m.get(k); - return v == null ? def : v.toString(); - } - - private static long num(Map m, String k, long def) { - Object v = m.get(k); - return v == null ? def : (long) ((Number) v).doubleValue(); + return new Config( + host, + port, + psk, + json.getLong("timestampWindowMs", 30_000L), + json.getLong("pendingTimeoutMs", 10_000L)); } } diff --git a/server/src/main/java/io/icybear/redapricot/ControlSession.java b/server/src/main/java/io/icybear/redapricot/ControlSession.java index 9a51051..b3715b5 100644 --- a/server/src/main/java/io/icybear/redapricot/ControlSession.java +++ b/server/src/main/java/io/icybear/redapricot/ControlSession.java @@ -3,24 +3,22 @@ package io.icybear.redapricot; import io.icybear.redapricot.net.EncryptedFrames; import io.icybear.redapricot.util.ProtoReader; import io.icybear.redapricot.util.ProtoWriter; +import lombok.RequiredArgsConstructor; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; /** * An authenticated control session (Magic 0x01). Carries pattern registrations * and control requests; never tunnels game data. */ +@RequiredArgsConstructor public final class ControlSession { - private static final System.Logger LOG = System.getLogger("redapricot.control"); + private static final Logger LOG = LogManager.getLogger("redapricot.control"); private final Hub hub; private final EncryptedFrames frames; private final String id; - public ControlSession(Hub hub, EncryptedFrames frames, String id) { - this.hub = hub; - this.frames = frames; - this.id = id; - } - public String id() { return id; } public void onFrame(byte[] payload) { @@ -40,10 +38,8 @@ public final class ControlSession { long nonce = r.readI64(); sendPong(nonce); } - case Protocol.FRAME_ERROR -> LOG.log(System.Logger.Level.WARNING, - "control " + id + " error: " + r.readString()); - default -> LOG.log(System.Logger.Level.WARNING, - "control " + id + " unknown message type " + type); + case Protocol.FRAME_ERROR -> LOG.warn("control {} error: {}", id, r.readString()); + default -> LOG.warn("control {} unknown message type {}", id, type); } } @@ -56,8 +52,7 @@ public final class ControlSession { .u16(playerPort) .toBytes(); frames.send(msg); - LOG.log(System.Logger.Level.INFO, - "control-request pattern=" + pattern + " player=" + playerIp + ":" + playerPort); + LOG.info("control-request pattern={} player={}:{}", pattern, playerIp, playerPort); } private void sendRegisterAck(String pattern, int status) { @@ -74,6 +69,6 @@ public final class ControlSession { public void onClose() { hub.removeSession(this); - LOG.log(System.Logger.Level.INFO, "control session " + id + " closed"); + LOG.info("control session {} closed", id); } } diff --git a/server/src/main/java/io/icybear/redapricot/Hub.java b/server/src/main/java/io/icybear/redapricot/Hub.java index cd33ccb..845435f 100644 --- a/server/src/main/java/io/icybear/redapricot/Hub.java +++ b/server/src/main/java/io/icybear/redapricot/Hub.java @@ -3,6 +3,8 @@ package io.icybear.redapricot; import io.icybear.redapricot.crypto.Crypto; import io.icybear.redapricot.util.Hex; import io.vertx.core.Vertx; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.nio.charset.StandardCharsets; import java.security.SecureRandom; @@ -17,7 +19,7 @@ import java.util.concurrent.ThreadLocalRandom; * concurrent maps are defensive. */ public final class Hub { - private static final System.Logger LOG = System.getLogger("redapricot.hub"); + private static final Logger LOG = LogManager.getLogger("redapricot.hub"); public final Vertx vertx; public final Config config; @@ -30,8 +32,8 @@ public final class Hub { public Hub(Vertx vertx, Config config) { this.vertx = vertx; this.config = config; - this.pskBytes = config.psk.getBytes(StandardCharsets.UTF_8); - this.pskAddress = Crypto.pskAddress(config.psk); + this.pskBytes = config.psk().getBytes(StandardCharsets.UTF_8); + this.pskAddress = Crypto.pskAddress(config.psk()); } // ---- pattern registry ---- @@ -39,7 +41,7 @@ public final class Hub { public void register(String pattern, ControlSession session) { String key = normalizeAddress(pattern); patterns.put(key, session); - LOG.log(System.Logger.Level.INFO, "registered pattern '" + key + "' -> " + session.id()); + LOG.info("registered pattern '{}' -> {}", key, session.id()); } public void unregister(String pattern, ControlSession session) { @@ -51,9 +53,16 @@ public final class Hub { return patterns.get(normalizeAddress(address)); } - /** Drop every pattern currently owned by a (closing) session. */ + /** Drop every pattern owned by a (closing) session, plus any players still pending for it. */ public void removeSession(ControlSession session) { patterns.entrySet().removeIf(e -> e.getValue() == session); + pending.values().removeIf(p -> { + if (p.getOwner() != session) return false; + if (p.getTimerId() >= 0) vertx.cancelTimer(p.getTimerId()); + p.getSocket().close(); + LOG.info("dropping pending player {} (control session {} closed)", p.getCidHex(), session.id()); + return true; + }); } // ---- pending players ---- @@ -65,26 +74,26 @@ public final class Hub { } public void addPending(PendingPlayer p) { - pending.put(p.cidHex, p); - p.timerId = vertx.setTimer(config.pendingTimeoutMs, id -> { - PendingPlayer removed = pending.remove(p.cidHex); + pending.put(p.getCidHex(), p); + p.setTimerId(vertx.setTimer(config.pendingTimeoutMs(), id -> { + PendingPlayer removed = pending.remove(p.getCidHex()); if (removed != null) { - LOG.log(System.Logger.Level.WARNING, "pending player " + p.cidHex + " timed out"); - removed.socket.close(); + LOG.warn("pending player {} timed out", p.getCidHex()); + removed.getSocket().close(); } - }); + })); } public PendingPlayer takePending(byte[] cid) { String hex = Hex.encode(cid); PendingPlayer p = pending.remove(hex); - if (p != null && p.timerId >= 0) vertx.cancelTimer(p.timerId); + if (p != null && p.getTimerId() >= 0) vertx.cancelTimer(p.getTimerId()); return p; } public void removePending(String cidHex) { PendingPlayer p = pending.remove(cidHex); - if (p != null && p.timerId >= 0) vertx.cancelTimer(p.timerId); + if (p != null && p.getTimerId() >= 0) vertx.cancelTimer(p.getTimerId()); } // ---- helpers ---- diff --git a/server/src/main/java/io/icybear/redapricot/HubConnection.java b/server/src/main/java/io/icybear/redapricot/HubConnection.java index a260dc2..33c590f 100644 --- a/server/src/main/java/io/icybear/redapricot/HubConnection.java +++ b/server/src/main/java/io/icybear/redapricot/HubConnection.java @@ -7,6 +7,8 @@ import io.icybear.redapricot.util.ProtoReader; import io.icybear.redapricot.util.VarInt; import io.vertx.core.buffer.Buffer; import io.vertx.core.net.NetSocket; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.util.concurrent.atomic.AtomicLong; @@ -16,7 +18,7 @@ import java.util.concurrent.atomic.AtomicLong; * connection as a player. See PROTOCOL.md §2 and §4. */ public final class HubConnection { - private static final System.Logger LOG = System.getLogger("redapricot.conn"); + private static final Logger LOG = LogManager.getLogger("redapricot.conn"); private static final AtomicLong SEQ = new AtomicLong(); private static final int MAX_HANDSHAKE = 8192; @@ -39,7 +41,10 @@ public final class HubConnection { public void start() { socket.handler(this::onRaw); - socket.closeHandler(v -> closeCleanup.run()); + socket.closeHandler(v -> { + if (frames != null) frames.markClosed(); + closeCleanup.run(); + }); socket.exceptionHandler(t -> socket.close()); } @@ -47,7 +52,7 @@ public final class HubConnection { if (dispatched) return; hs.appendBuffer(b); if (hs.length() > MAX_HANDSHAKE) { - LOG.log(System.Logger.Level.WARNING, id + " handshake too large; closing"); + LOG.warn("{} handshake too large; closing", id); socket.close(); return; } @@ -77,7 +82,7 @@ public final class HubConnection { try { dispatch(packet, afterHandshake); } catch (RuntimeException e) { - LOG.log(System.Logger.Level.WARNING, id + " handshake error: " + e); + LOG.warn("{} handshake error: {}", id, e); socket.close(); } } @@ -97,7 +102,7 @@ public final class HubConnection { if (intent == Protocol.INTENT_REDAPRICOT) { beginRedapricot(address, afterHandshake); } else if (intent == Protocol.INTENT_RESERVED) { - LOG.log(System.Logger.Level.INFO, id + " reserved intent 18; closing"); + LOG.info("{} reserved intent 18; closing", id); socket.close(); } else { handlePlayer(address); @@ -108,7 +113,7 @@ public final class HubConnection { private void beginRedapricot(String address, Buffer afterHandshake) { if (!address.equalsIgnoreCase(hub.pskAddress)) { - LOG.log(System.Logger.Level.WARNING, id + " bad PSK address; closing"); + LOG.warn("{} bad PSK address; closing", id); socket.close(); return; } @@ -127,15 +132,15 @@ public final class HubConnection { int magic = r.readUByte(); int randLen = r.readVarInt(); if (randLen < 8 || randLen > 64) { - LOG.log(System.Logger.Level.WARNING, id + " bad rekey randLen; closing"); + LOG.warn("{} bad rekey randLen; closing", id); frames.close(); return; } byte[] rand = r.readBytes(randLen); long ts = r.readI64(); long now = System.currentTimeMillis(); - if (Math.abs(now - ts) > hub.config.timestampWindowMs) { - LOG.log(System.Logger.Level.WARNING, id + " rekey timestamp outside window; closing"); + if (Math.abs(now - ts) > hub.config.timestampWindowMs()) { + LOG.warn("{} rekey timestamp outside window; closing", id); frames.close(); return; } @@ -158,14 +163,14 @@ public final class HubConnection { ControlSession session = new ControlSession(hub, frames, id); frames.setHandler(session::onFrame); closeCleanup = session::onClose; - LOG.log(System.Logger.Level.INFO, id + " control session established"); + LOG.info("{} control session established", id); } else if (magic == Protocol.MAGIC_WORKER) { WorkerConn worker = new WorkerConn(hub, frames, id); frames.setHandler(worker::onFrame); closeCleanup = worker::onClose; - LOG.log(System.Logger.Level.INFO, id + " worker conn established"); + LOG.info("{} worker conn established", id); } else { - LOG.log(System.Logger.Level.WARNING, id + " bad magic " + magic + "; closing"); + LOG.warn("{} bad magic {}; closing", id, magic); frames.close(); } } @@ -176,7 +181,7 @@ public final class HubConnection { String pattern = Hub.normalizeAddress(address); ControlSession session = hub.match(address); if (session == null) { - LOG.log(System.Logger.Level.INFO, id + " no route for '" + pattern + "'; closing"); + LOG.info("{} no route for '{}'; closing", id, pattern); socket.close(); return; } @@ -188,12 +193,11 @@ public final class HubConnection { socket.pause(); Buffer buffered = hs.copy(); // handshake + any pipelined bytes, forwarded verbatim - PendingPlayer p = new PendingPlayer(cid, cidHex, socket, buffered, pattern, ip, port); + PendingPlayer p = new PendingPlayer(cid, cidHex, socket, buffered, pattern, ip, port, session); hub.addPending(p); closeCleanup = () -> hub.removePending(cidHex); session.sendControlRequest(cid, pattern, ip, port); - LOG.log(System.Logger.Level.INFO, - id + " player " + ip + ":" + port + " matched '" + pattern + "' cid=" + cidHex); + LOG.info("{} player {}:{} matched '{}' cid={}", id, ip, port, pattern, cidHex); } } diff --git a/server/src/main/java/io/icybear/redapricot/HubServer.java b/server/src/main/java/io/icybear/redapricot/HubServer.java index 12b8c2d..ff3591d 100644 --- a/server/src/main/java/io/icybear/redapricot/HubServer.java +++ b/server/src/main/java/io/icybear/redapricot/HubServer.java @@ -4,25 +4,25 @@ import io.vertx.core.AbstractVerticle; import io.vertx.core.Promise; import io.vertx.core.net.NetServer; import io.vertx.core.net.NetServerOptions; +import lombok.RequiredArgsConstructor; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; /** Vert.x verticle that accepts every inbound connection on the hub port. */ +@RequiredArgsConstructor public final class HubServer extends AbstractVerticle { - private static final System.Logger LOG = System.getLogger("redapricot.server"); + private static final Logger LOG = LogManager.getLogger("redapricot.server"); private final Config config; private Hub hub; - public HubServer(Config config) { - this.config = config; - } - @Override public void start(Promise startPromise) { hub = new Hub(vertx, config); NetServerOptions opts = new NetServerOptions() - .setHost(config.host) - .setPort(config.port) + .setHost(config.host()) + .setPort(config.port()) .setTcpNoDelay(true) .setReuseAddress(true); @@ -30,9 +30,8 @@ public final class HubServer extends AbstractVerticle { server.connectHandler(socket -> new HubConnection(hub, socket).start()); server.listen().onComplete(ar -> { if (ar.succeeded()) { - LOG.log(System.Logger.Level.INFO, - "redapricot hub listening on " + config.host + ":" + ar.result().actualPort()); - LOG.log(System.Logger.Level.INFO, "PSK handshake address: " + hub.pskAddress); + LOG.info("redapricot hub listening on {}:{}", config.host(), ar.result().actualPort()); + LOG.info("PSK handshake address: {}", hub.pskAddress); startPromise.complete(); } else { startPromise.fail(ar.cause()); diff --git a/server/src/main/java/io/icybear/redapricot/Main.java b/server/src/main/java/io/icybear/redapricot/Main.java index 177f66a..2bc7198 100644 --- a/server/src/main/java/io/icybear/redapricot/Main.java +++ b/server/src/main/java/io/icybear/redapricot/Main.java @@ -2,12 +2,22 @@ package io.icybear.redapricot; import io.vertx.core.Vertx; import io.vertx.core.VertxOptions; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.nio.file.Path; /** Entry point: {@code java -jar redapricot-server.jar }. */ public final class Main { - private static final System.Logger LOG = System.getLogger("redapricot.main"); + static { + // Must run before any JUL logger (Netty's fallback) or Vert.x logger is created, so keep + // these as the very first thing the class does. Covers `java -jar`, which has no JVM args. + System.setProperty("java.util.logging.manager", "org.apache.logging.log4j.jul.LogManager"); + System.setProperty("vertx.logger-delegate-factory-class-name", + "io.vertx.core.logging.Log4j2LogDelegateFactory"); + } + + private static final Logger LOG = LogManager.getLogger("redapricot.main"); public static void main(String[] args) throws Exception { if (args.length < 1) { @@ -21,7 +31,7 @@ public final class Main { Vertx vertx = Vertx.vertx(new VertxOptions()); vertx.deployVerticle(new HubServer(config)).onComplete(ar -> { if (ar.failed()) { - LOG.log(System.Logger.Level.ERROR, "failed to start hub", ar.cause()); + LOG.error("failed to start hub", ar.cause()); vertx.close(); System.exit(1); } diff --git a/server/src/main/java/io/icybear/redapricot/PendingPlayer.java b/server/src/main/java/io/icybear/redapricot/PendingPlayer.java index cc581ee..00c1b7d 100644 --- a/server/src/main/java/io/icybear/redapricot/PendingPlayer.java +++ b/server/src/main/java/io/icybear/redapricot/PendingPlayer.java @@ -2,26 +2,23 @@ package io.icybear.redapricot; import io.vertx.core.buffer.Buffer; import io.vertx.core.net.NetSocket; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.Setter; /** A player connection awaiting a worker-conn takeover, keyed by CID. */ +@Getter +@RequiredArgsConstructor public final class PendingPlayer { - public final byte[] cid; - public final String cidHex; - public final NetSocket socket; - public final Buffer buffered; // raw bytes already read from the player (handshake + pipelined) - public final String pattern; - public final String playerIp; - public final int playerPort; - public long timerId = -1; + private final byte[] cid; + private final String cidHex; + private final NetSocket socket; + private final Buffer buffered; // raw bytes already read from the player (handshake + pipelined) + private final String pattern; + private final String playerIp; + private final int playerPort; + private final ControlSession owner; // control session this player was routed to - public PendingPlayer(byte[] cid, String cidHex, NetSocket socket, Buffer buffered, - String pattern, String playerIp, int playerPort) { - this.cid = cid; - this.cidHex = cidHex; - this.socket = socket; - this.buffered = buffered; - this.pattern = pattern; - this.playerIp = playerIp; - this.playerPort = playerPort; - } + @Setter + private long timerId = -1; } diff --git a/server/src/main/java/io/icybear/redapricot/WorkerConn.java b/server/src/main/java/io/icybear/redapricot/WorkerConn.java index f84fa2d..28a229b 100644 --- a/server/src/main/java/io/icybear/redapricot/WorkerConn.java +++ b/server/src/main/java/io/icybear/redapricot/WorkerConn.java @@ -5,27 +5,32 @@ import io.icybear.redapricot.util.ProtoReader; import io.icybear.redapricot.util.ProtoWriter; import io.vertx.core.buffer.Buffer; import io.vertx.core.net.NetSocket; +import lombok.RequiredArgsConstructor; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; /** * An authenticated worker connection (Magic 0x02). Multiplexes many player * streams; the client opens streams via SYN(CID) to take over pending players. */ +@RequiredArgsConstructor public final class WorkerConn { - private static final System.Logger LOG = System.getLogger("redapricot.worker"); + private static final Logger LOG = LogManager.getLogger("redapricot.worker"); private final Hub hub; private final EncryptedFrames frames; private final String id; private final Map streams = new HashMap<>(); - public WorkerConn(Hub hub, EncryptedFrames frames, String id) { - this.hub = hub; - this.frames = frames; - this.id = id; - } + // Backpressure for the single shared worker socket, arbitrated across all streams. + private final Set upstreamPaused = new HashSet<>(); // players parked until the worker write queue drains + private final Set downstreamBlocked = new HashSet<>(); // sids whose player write queue is full; non-empty => worker read paused + private boolean workerDrainArmed = false; // whether the worker socket's single drainHandler is set public void onFrame(byte[] payload) { ProtoReader r = new ProtoReader(payload); @@ -33,22 +38,21 @@ public final class WorkerConn { int sid = r.readVarInt(); switch (type) { case Protocol.MUX_SYN -> handleSyn(sid, r.readBytes(Protocol.CID_LEN)); - case Protocol.MUX_DATA -> handleData(sid, r.readBytes(r.remaining())); + case Protocol.MUX_DATA -> handleData(sid, r.readBuffer(r.remaining())); case Protocol.MUX_FIN, Protocol.MUX_RST -> closeStream(sid); - case Protocol.FRAME_ERROR -> LOG.log(System.Logger.Level.WARNING, - "worker " + id + " error frame"); - default -> LOG.log(System.Logger.Level.WARNING, "worker " + id + " unknown mux type " + type); + case Protocol.FRAME_ERROR -> LOG.warn("worker {} error frame", id); + default -> LOG.warn("worker {} unknown mux type {}", id, type); } } private void handleSyn(int sid, byte[] cid) { PendingPlayer p = hub.takePending(cid); if (p == null) { - LOG.log(System.Logger.Level.WARNING, "worker " + id + " SYN for unknown CID"); + LOG.warn("worker {} SYN for unknown CID", id); sendRst(sid); return; } - NetSocket player = p.socket; + NetSocket player = p.getSocket(); streams.put(sid, player); // From now on the player socket belongs to this stream. @@ -56,36 +60,66 @@ public final class WorkerConn { sendData(sid, buf.getBytes()); if (frames.writeQueueFull()) { player.pause(); - frames.socket().drainHandler(v -> player.resume()); + upstreamPaused.add(player); + armWorkerDrain(); } }); - player.closeHandler(v -> { - if (streams.remove(sid) != null) sendFin(sid); - }); - player.exceptionHandler(t -> { - if (streams.remove(sid) != null) sendFin(sid); - }); + player.closeHandler(v -> onPlayerGone(sid, player)); + player.exceptionHandler(t -> onPlayerGone(sid, player)); // Forward the buffered handshake (and any pipelined bytes), then resume. - sendData(sid, p.buffered.getBytes()); + sendData(sid, p.getBuffered().getBytes()); player.resume(); - LOG.log(System.Logger.Level.INFO, - "worker " + id + " stream " + sid + " bound to " + p.pattern); + LOG.info("worker {} stream {} bound to {}", id, sid, p.getPattern()); } - private void handleData(int sid, byte[] data) { + private void handleData(int sid, Buffer data) { NetSocket player = streams.get(sid); if (player == null) return; - player.write(Buffer.buffer(data)); - if (player.writeQueueFull()) { - frames.socket().pause(); - player.drainHandler(v -> frames.socket().resume()); + player.write(data); + // A slow player pauses the shared worker read side; the block set lets us resume only + // once every blocked player has drained (and release it if a player disconnects meanwhile). + if (player.writeQueueFull() && downstreamBlocked.add(sid)) { + if (downstreamBlocked.size() == 1) frames.socket().pause(); + player.drainHandler(v -> unblockDownstream(sid)); } } private void closeStream(int sid) { NetSocket player = streams.remove(sid); - if (player != null) player.close(); + unblockDownstream(sid); + if (player != null) { + upstreamPaused.remove(player); + player.close(); + } + } + + /** Register (once) the shared worker socket's single drain handler; on drain, wake every parked player. */ + private void armWorkerDrain() { + if (workerDrainArmed) return; + workerDrainArmed = true; + frames.socket().drainHandler(v -> { + workerDrainArmed = false; + if (upstreamPaused.isEmpty()) return; + NetSocket[] parked = upstreamPaused.toArray(new NetSocket[0]); + upstreamPaused.clear(); + for (NetSocket pl : parked) pl.resume(); + }); + } + + /** A blocked player drained or vanished: drop its downstream block, resuming the shared worker read side when none remain. */ + private void unblockDownstream(int sid) { + if (downstreamBlocked.remove(sid) && downstreamBlocked.isEmpty()) { + frames.socket().resume(); + } + } + + /** The player side of a stream vanished: drop it from every table, release any backpressure it held, and FIN the peer if still live. */ + private void onPlayerGone(int sid, NetSocket player) { + boolean wasLive = streams.remove(sid) != null; + upstreamPaused.remove(player); + unblockDownstream(sid); + if (wasLive) sendFin(sid); } private void sendData(int sid, byte[] data) { @@ -103,6 +137,8 @@ public final class WorkerConn { public void onClose() { for (NetSocket player : streams.values()) player.close(); streams.clear(); - LOG.log(System.Logger.Level.INFO, "worker " + id + " closed"); + downstreamBlocked.clear(); + upstreamPaused.clear(); + LOG.info("worker {} closed", id); } } diff --git a/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java b/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java index 444db0d..4bdc135 100644 --- a/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java +++ b/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java @@ -101,4 +101,9 @@ public final class EncryptedFrames { closed = true; socket.close(); } + + /** Mark the transport closed after the peer closed the socket, without initiating another close. */ + public void markClosed() { + closed = true; + } } diff --git a/server/src/main/java/io/icybear/redapricot/util/Json.java b/server/src/main/java/io/icybear/redapricot/util/Json.java deleted file mode 100644 index 0eb0c7c..0000000 --- a/server/src/main/java/io/icybear/redapricot/util/Json.java +++ /dev/null @@ -1,151 +0,0 @@ -package io.icybear.redapricot.util; - -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; - -/** - * Minimal, dependency-free JSON parser sufficient for redapricot config files. - * Produces Map<String,Object>, List<Object>, String, Double, Boolean, null. - */ -public final class Json { - private final String s; - private int i; - - private Json(String s) { this.s = s; } - - public static Object parse(String text) { - Json p = new Json(text); - p.ws(); - Object v = p.value(); - p.ws(); - if (p.i != p.s.length()) throw new IllegalArgumentException("trailing JSON at " + p.i); - return v; - } - - @SuppressWarnings("unchecked") - public static Map parseObject(String text) { - Object v = parse(text); - if (!(v instanceof Map)) throw new IllegalArgumentException("expected JSON object"); - return (Map) v; - } - - private Object value() { - char c = peek(); - return switch (c) { - case '{' -> object(); - case '[' -> array(); - case '"' -> string(); - case 't', 'f' -> bool(); - case 'n' -> nul(); - default -> number(); - }; - } - - private Map object() { - expect('{'); - Map m = new LinkedHashMap<>(); - ws(); - if (peek() == '}') { i++; return m; } - while (true) { - ws(); - String key = string(); - ws(); - expect(':'); - ws(); - m.put(key, value()); - ws(); - char c = next(); - if (c == '}') return m; - if (c != ',') throw err("expected , or }"); - } - } - - private List array() { - expect('['); - List a = new ArrayList<>(); - ws(); - if (peek() == ']') { i++; return a; } - while (true) { - ws(); - a.add(value()); - ws(); - char c = next(); - if (c == ']') return a; - if (c != ',') throw err("expected , or ]"); - } - } - - private String string() { - expect('"'); - StringBuilder sb = new StringBuilder(); - while (true) { - char c = next(); - if (c == '"') return sb.toString(); - if (c == '\\') { - char e = next(); - switch (e) { - case '"' -> sb.append('"'); - case '\\' -> sb.append('\\'); - case '/' -> sb.append('/'); - case 'b' -> sb.append('\b'); - case 'f' -> sb.append('\f'); - case 'n' -> sb.append('\n'); - case 'r' -> sb.append('\r'); - case 't' -> sb.append('\t'); - case 'u' -> { - int cp = Integer.parseInt(s.substring(i, i + 4), 16); - i += 4; - sb.append((char) cp); - } - default -> throw err("bad escape"); - } - } else { - sb.append(c); - } - } - } - - private Object number() { - int start = i; - while (i < s.length() && "+-0123456789.eE".indexOf(s.charAt(i)) >= 0) i++; - String num = s.substring(start, i); - if (num.isEmpty()) throw err("bad value"); - if (num.contains(".") || num.contains("e") || num.contains("E")) return Double.parseDouble(num); - return Double.parseDouble(num); // keep numbers as Double uniformly - } - - private Boolean bool() { - if (s.startsWith("true", i)) { i += 4; return Boolean.TRUE; } - if (s.startsWith("false", i)) { i += 5; return Boolean.FALSE; } - throw err("bad literal"); - } - - private Object nul() { - if (s.startsWith("null", i)) { i += 4; return null; } - throw err("bad literal"); - } - - private void ws() { - while (i < s.length() && Character.isWhitespace(s.charAt(i))) i++; - } - - private char peek() { - if (i >= s.length()) throw err("unexpected end"); - return s.charAt(i); - } - - private char next() { - if (i >= s.length()) throw err("unexpected end"); - return s.charAt(i++); - } - - private void expect(char c) { - if (next() != c) throw err("expected " + c); - } - - private IllegalArgumentException err(String msg) { - return new IllegalArgumentException("JSON: " + msg + " at index " + i); - } -} diff --git a/server/src/main/java/io/icybear/redapricot/util/ProtoReader.java b/server/src/main/java/io/icybear/redapricot/util/ProtoReader.java index 20761c1..4163ebf 100644 --- a/server/src/main/java/io/icybear/redapricot/util/ProtoReader.java +++ b/server/src/main/java/io/icybear/redapricot/util/ProtoReader.java @@ -1,62 +1,71 @@ package io.icybear.redapricot.util; +import io.vertx.core.buffer.Buffer; + import java.nio.charset.StandardCharsets; -/** Cursor-based reader for redapricot/Minecraft primitive types over a byte array. */ +/** Cursor-based reader for redapricot/Minecraft primitive types over a Vert.x {@link Buffer}. */ public final class ProtoReader { - private final byte[] buf; + private final Buffer buf; private int pos; - private final int end; - public ProtoReader(byte[] buf) { this(buf, 0, buf.length); } - - public ProtoReader(byte[] buf, int off, int len) { + public ProtoReader(Buffer buf) { this.buf = buf; - this.pos = off; - this.end = off + len; } - public int remaining() { return end - pos; } + /** Convenience for callers holding a raw payload (e.g. decrypted frame bytes). */ + public ProtoReader(byte[] bytes) { + this(Buffer.buffer(bytes)); + } + + public int remaining() { return buf.length() - pos; } public int readUByte() { - if (pos >= end) throw new IllegalStateException("underflow"); - return buf[pos++] & 0xFF; + if (pos >= buf.length()) throw new IllegalStateException("underflow"); + return buf.getUnsignedByte(pos++); } public int readVarInt() { - int value = 0; - int shift = 0; - while (true) { - int b = readUByte(); - value |= (b & 0x7F) << shift; - if ((b & 0x80) == 0) return value; - shift += 7; - if (shift >= 32) throw new IllegalArgumentException("VarInt too big"); - } + VarInt.Read r = VarInt.tryRead(buf, pos); + if (r == null) throw new IllegalStateException("underflow"); + pos += r.size(); + return r.value(); } public int readU16() { - int hi = readUByte(); - int lo = readUByte(); - return (hi << 8) | lo; + if (remaining() < 2) throw new IllegalStateException("underflow"); + int v = buf.getUnsignedShort(pos); + pos += 2; + return v; } public long readI64() { - long v = 0; - for (int i = 0; i < 8; i++) v = (v << 8) | readUByte(); + if (remaining() < 8) throw new IllegalStateException("underflow"); + long v = buf.getLong(pos); + pos += 8; return v; } public byte[] readBytes(int n) { - if (n < 0 || n > remaining()) throw new IllegalStateException("bad length " + n); - byte[] out = new byte[n]; - System.arraycopy(buf, pos, out, 0, n); + checkLen(n); + byte[] out = buf.getBytes(pos, pos + n); pos += n; return out; } + /** Read {@code n} bytes as a Buffer slice — avoids the intermediate byte[] on data-forwarding paths. */ + public Buffer readBuffer(int n) { + checkLen(n); + Buffer slice = buf.getBuffer(pos, pos + n); + pos += n; + return slice; + } + public String readString() { - int len = readVarInt(); - return new String(readBytes(len), StandardCharsets.UTF_8); + return new String(readBytes(readVarInt()), StandardCharsets.UTF_8); + } + + private void checkLen(int n) { + if (n < 0 || n > remaining()) throw new IllegalStateException("bad length " + n); } } diff --git a/server/src/main/resources/log4j2.xml b/server/src/main/resources/log4j2.xml new file mode 100644 index 0000000..6ada93e --- /dev/null +++ b/server/src/main/resources/log4j2.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + +