cloud/redapricot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: logging and improved multiplex handling

Browse Source
This commit is contained in:
iceBear67
2026-07-15 16:43:43 +08:00
parent 6e0d7ec33f
commit ea33744ec9
13 changed files with 238 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/build.gradle.kts
+7
View File
@@ -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 {
server/src/main/java/io/icybear/redapricot/Config.java
+19 -32
View File
@@ -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<String, Object> 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<String, Object> m, String k, String def) {
Object v = m.get(k);
return v == null ? def : v.toString();
}
private static long num(Map<String, Object> 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));
}
}
server/src/main/java/io/icybear/redapricot/ControlSession.java
+9 -14
View File
@@ -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);
}
}
server/src/main/java/io/icybear/redapricot/Hub.java
+22 -13
View File
@@ -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 ----
server/src/main/java/io/icybear/redapricot/HubConnection.java
+20 -16
View File
@@ -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);
}
}
server/src/main/java/io/icybear/redapricot/HubServer.java
+9 -10
View File
@@ -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<Void> 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());
server/src/main/java/io/icybear/redapricot/Main.java
+12 -2
View File
@@ -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 <config.json>}. */
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);
}
server/src/main/java/io/icybear/redapricot/PendingPlayer.java
+15 -18
View File
@@ -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;
}
server/src/main/java/io/icybear/redapricot/WorkerConn.java
+65 -29
View File
@@ -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<Integer, NetSocket> 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<NetSocket> upstreamPaused = new HashSet<>(); // players parked until the worker write queue drains
private final Set<Integer> 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);
}
}
server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java
+5
View File
@@ -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;
}
}
server/src/main/java/io/icybear/redapricot/util/Json.java
-151
View File
@@ -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&lt;String,Object&gt;, List&lt;Object&gt;, 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<String, Object> parseObject(String text) {
Object v = parse(text);
if (!(v instanceof Map)) throw new IllegalArgumentException("expected JSON object");
return (Map<String, Object>) 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<String, Object> object() {
expect('{');
Map<String, Object> 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<Object> array() {
expect('[');
List<Object> 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);
}
}
server/src/main/java/io/icybear/redapricot/util/ProtoReader.java
+39 -30
View File
@@ -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);
}
}
server/src/main/resources/log4j2.xml
+16
View File
@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout
pattern="%d{HH:mm:ss.SSS} %highlight{%-5level}{FATAL=red bold, ERROR=red, WARN=yellow, INFO=green, DEBUG=cyan, TRACE=blue} %style{%logger{1}}{cyan} %msg%n%throwable"/>
</Console>
</Appenders>
<Loggers>
<Logger name="io.netty" level="WARN"/> <!-- quiet Netty internals -->
<Logger name="io.vertx" level="INFO"/>
<Root level="INFO">
<AppenderRef ref="Console"/>
</Root>
</Loggers>
</Configuration>