Phase 1: forge-1_12_2 WS connection and column delta capture
Adds a hand-rolled RFC 6455 WS client and minimal JSON codec to common/ (no third-party deps, keeping legacy ForgeGradle's classpath untouched), a DefaultBackendConnection implementing hello/hello_ack auth and reconnect with backoff, and Forge1122ChunkAdapter deriving top-of-column state from the vanilla heightmap for both initial chunk backfill and event-driven deltas (block break/place, deferred one tick to read post-mutation state). MCMapperMod wires it up with config-driven backendUrl/serverToken and a tick-based flush batch. Verified against a live MCMapper-Backend api instance (real WS handshake, hello_ack, and 256-column batch landing correctly in Postgres).
This commit is contained in:
@@ -0,0 +1,206 @@
|
||||
package com.octoturge.mcmapper.common;
|
||||
|
||||
import com.octoturge.mcmapper.common.json.MiniJson;
|
||||
import com.octoturge.mcmapper.common.protocol.DeltaEvent;
|
||||
import com.octoturge.mcmapper.common.protocol.LinkRequest;
|
||||
import com.octoturge.mcmapper.common.ws.SimpleWebSocketClient;
|
||||
|
||||
import java.net.URI;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* Wire protocol (mod -> api / api -> mod), one JSON object per WS text frame — mirrored in
|
||||
* MCMapper-Backend's {@code api/src/ws-gateway.ts}:
|
||||
*
|
||||
* <pre>
|
||||
* mod -> api {"type":"hello","token":"<serverToken>"}
|
||||
* api -> mod {"type":"hello_ack","ok":true,"serverId":"<uuid>"}
|
||||
* {"type":"hello_ack","ok":false,"error":"..."}
|
||||
*
|
||||
* mod -> api {"type":"columns","dimension":0,"columns":[{"x":..,"z":..,"height":..,"blockId":..,"blockMeta":..}]}
|
||||
* </pre>
|
||||
*
|
||||
* A "columns" message doubles as both initial backfill (one message per loaded chunk) and live
|
||||
* deltas (one message per flush tick) — see {@link DeltaEvent}'s javadoc for how a
|
||||
* {@code List<DeltaEvent>} maps onto it. {@code dimension} in {@link DeltaEvent} is a string so
|
||||
* NeoForge's namespaced dimension keys (Phase 10) can reuse this type; pre-Flattening leaves
|
||||
* (1.7.10/1.12.2) always populate it with a stringified vanilla dimension id (e.g. {@code "0"}),
|
||||
* which is what lets this class safely {@code Integer.parseInt} it for the wire message.
|
||||
*
|
||||
* No offline queue: deltas sent while disconnected are dropped rather than buffered — the
|
||||
* periodic reconciliation sweep (not yet built, see plan's Phase 7) is what's meant to catch
|
||||
* whatever a disconnect window missed, so buffering here would be solving the same problem twice.
|
||||
*
|
||||
* This class avoids a third-party JSON/WS library entirely (see {@link SimpleWebSocketClient}
|
||||
* and {@link MiniJson}'s javadoc) to keep the mod's classpath free of anything that would need
|
||||
* shading through legacy ForgeGradle.
|
||||
*/
|
||||
public class DefaultBackendConnection implements BackendConnection {
|
||||
private final Consumer<String> logInfo;
|
||||
private final Consumer<String> logWarn;
|
||||
|
||||
private volatile SimpleWebSocketClient ws;
|
||||
private volatile boolean serverReady = false;
|
||||
private volatile boolean shouldReconnect = false;
|
||||
private volatile long reconnectDelayMs = 1000;
|
||||
private static final long MAX_RECONNECT_DELAY_MS = 30_000;
|
||||
|
||||
private String url;
|
||||
private String serverToken;
|
||||
|
||||
public DefaultBackendConnection(Consumer<String> logInfo, Consumer<String> logWarn) {
|
||||
this.logInfo = logInfo;
|
||||
this.logWarn = logWarn;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void connect(String url, String serverToken) {
|
||||
this.url = url;
|
||||
this.serverToken = serverToken;
|
||||
this.shouldReconnect = true;
|
||||
this.reconnectDelayMs = 1000;
|
||||
doConnect();
|
||||
}
|
||||
|
||||
private void doConnect() {
|
||||
SimpleWebSocketClient client = new SimpleWebSocketClient(new SimpleWebSocketClient.Listener() {
|
||||
@Override
|
||||
public void onOpen() {
|
||||
reconnectDelayMs = 1000;
|
||||
sendRaw("hello", map("type", "hello", "token", serverToken));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onText(String message) {
|
||||
handleMessage(message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onClose(int code, String reason) {
|
||||
serverReady = false;
|
||||
logWarn.accept("backend connection closed (" + code + " " + reason + ")");
|
||||
if (shouldReconnect) scheduleReconnect();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Exception e) {
|
||||
logWarn.accept("backend connection error: " + e);
|
||||
}
|
||||
});
|
||||
this.ws = client;
|
||||
try {
|
||||
client.connect(URI.create(url));
|
||||
} catch (Exception e) {
|
||||
logWarn.accept("failed to connect to backend at " + url + ": " + e);
|
||||
if (shouldReconnect) scheduleReconnect();
|
||||
}
|
||||
}
|
||||
|
||||
private void scheduleReconnect() {
|
||||
long delay = reconnectDelayMs;
|
||||
reconnectDelayMs = Math.min(reconnectDelayMs * 2, MAX_RECONNECT_DELAY_MS);
|
||||
Thread t = new Thread(() -> {
|
||||
try {
|
||||
Thread.sleep(delay);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return;
|
||||
}
|
||||
if (shouldReconnect) doConnect();
|
||||
}, "mcmapper-ws-reconnect");
|
||||
t.setDaemon(true);
|
||||
t.start();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private void handleMessage(String message) {
|
||||
Object parsed;
|
||||
try {
|
||||
parsed = MiniJson.parse(message);
|
||||
} catch (Exception e) {
|
||||
logWarn.accept("received malformed message from backend: " + e);
|
||||
return;
|
||||
}
|
||||
if (!(parsed instanceof Map)) return;
|
||||
Map<String, Object> obj = (Map<String, Object>) parsed;
|
||||
String type = String.valueOf(obj.get("type"));
|
||||
|
||||
if ("hello_ack".equals(type)) {
|
||||
boolean ok = Boolean.TRUE.equals(obj.get("ok"));
|
||||
if (ok) {
|
||||
serverReady = true;
|
||||
logInfo.accept("authenticated with backend as server " + obj.get("serverId"));
|
||||
} else {
|
||||
serverReady = false;
|
||||
logWarn.accept("backend rejected connection: " + obj.get("error"));
|
||||
shouldReconnect = false; // an invalid token won't fix itself by retrying
|
||||
}
|
||||
} else if ("error".equals(type)) {
|
||||
logWarn.accept("backend reported error: " + obj.get("error"));
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendDeltas(List<DeltaEvent> deltas) {
|
||||
if (deltas.isEmpty() || !serverReady) return;
|
||||
|
||||
// Group by dimension — the wire message is single-dimension, see class javadoc.
|
||||
Set<String> dimensions = new LinkedHashSet<>();
|
||||
for (DeltaEvent e : deltas) dimensions.add(e.dimension);
|
||||
|
||||
for (String dimension : dimensions) {
|
||||
List<Object> columns = new ArrayList<>();
|
||||
for (DeltaEvent e : deltas) {
|
||||
if (!e.dimension.equals(dimension)) continue;
|
||||
Map<String, Object> col = new LinkedHashMap<>();
|
||||
col.put("x", (double) e.x);
|
||||
col.put("z", (double) e.z);
|
||||
col.put("height", (double) e.y);
|
||||
col.put("blockId", (double) (e.blockStateId >> 4));
|
||||
col.put("blockMeta", (double) (e.blockStateId & 0xF));
|
||||
columns.add(col);
|
||||
}
|
||||
Map<String, Object> msg = new LinkedHashMap<>();
|
||||
msg.put("type", "columns");
|
||||
msg.put("dimension", (double) Integer.parseInt(dimension));
|
||||
msg.put("columns", columns);
|
||||
sendRaw("columns", msg);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendLinkRequest(LinkRequest request) {
|
||||
// The `/mcmapper link` flow (Phase 3) isn't wired up yet — nothing calls this in Phase 1.
|
||||
logWarn.accept("sendLinkRequest called before Phase 3's link flow is implemented — ignoring");
|
||||
}
|
||||
|
||||
private void sendRaw(String label, Map<String, Object> message) {
|
||||
SimpleWebSocketClient client = ws;
|
||||
if (client == null || !client.isOpen()) return;
|
||||
try {
|
||||
client.sendText(MiniJson.writeObject(message));
|
||||
} catch (Exception e) {
|
||||
logWarn.accept("failed to send '" + label + "' to backend: " + e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void disconnect() {
|
||||
shouldReconnect = false;
|
||||
serverReady = false;
|
||||
SimpleWebSocketClient client = ws;
|
||||
if (client != null) client.close();
|
||||
}
|
||||
|
||||
private static Map<String, Object> map(Object... kv) {
|
||||
Map<String, Object> m = new LinkedHashMap<>();
|
||||
for (int i = 0; i < kv.length; i += 2) m.put((String) kv[i], kv[i + 1]);
|
||||
return m;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
package com.octoturge.mcmapper.common.json;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* A minimal JSON reader/writer, hand-rolled so the mod doesn't need to shade in a JSON library
|
||||
* (Gson/Jackson) through legacy ForgeGradle's classpath — see BackendConnection's javadoc for
|
||||
* why the mod side stays dependency-free wherever reasonable. Handles the full JSON grammar
|
||||
* (objects, arrays, strings, numbers, booleans, null) since the extra code over a "flat objects
|
||||
* only" parser is small and this is reused for every inbound/outbound wire message.
|
||||
*/
|
||||
public final class MiniJson {
|
||||
private MiniJson() {
|
||||
}
|
||||
|
||||
// ---- Writing ----
|
||||
|
||||
public static String writeObject(Map<String, ?> fields) {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
writeValue(sb, fields);
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static void writeValue(StringBuilder sb, Object value) {
|
||||
if (value == null) {
|
||||
sb.append("null");
|
||||
} else if (value instanceof String) {
|
||||
writeString(sb, (String) value);
|
||||
} else if (value instanceof Number || value instanceof Boolean) {
|
||||
sb.append(value);
|
||||
} else if (value instanceof Map) {
|
||||
sb.append('{');
|
||||
boolean first = true;
|
||||
for (Map.Entry<String, ?> entry : ((Map<String, ?>) value).entrySet()) {
|
||||
if (!first) sb.append(',');
|
||||
first = false;
|
||||
writeString(sb, entry.getKey());
|
||||
sb.append(':');
|
||||
writeValue(sb, entry.getValue());
|
||||
}
|
||||
sb.append('}');
|
||||
} else if (value instanceof List) {
|
||||
sb.append('[');
|
||||
boolean first = true;
|
||||
for (Object item : (List<?>) value) {
|
||||
if (!first) sb.append(',');
|
||||
first = false;
|
||||
writeValue(sb, item);
|
||||
}
|
||||
sb.append(']');
|
||||
} else {
|
||||
throw new IllegalArgumentException("Unsupported JSON value type: " + value.getClass());
|
||||
}
|
||||
}
|
||||
|
||||
private static void writeString(StringBuilder sb, String s) {
|
||||
sb.append('"');
|
||||
for (int i = 0; i < s.length(); i++) {
|
||||
char c = s.charAt(i);
|
||||
switch (c) {
|
||||
case '"': sb.append("\\\""); break;
|
||||
case '\\': sb.append("\\\\"); break;
|
||||
case '\n': sb.append("\\n"); break;
|
||||
case '\r': sb.append("\\r"); break;
|
||||
case '\t': sb.append("\\t"); break;
|
||||
default:
|
||||
if (c < 0x20) {
|
||||
sb.append(String.format("\\u%04x", (int) c));
|
||||
} else {
|
||||
sb.append(c);
|
||||
}
|
||||
}
|
||||
}
|
||||
sb.append('"');
|
||||
}
|
||||
|
||||
// ---- Reading ----
|
||||
|
||||
/** Parses one JSON value (typically an object) from the start of {@code text}. */
|
||||
public static Object parse(String text) {
|
||||
Parser parser = new Parser(text);
|
||||
Object value = parser.parseValue();
|
||||
parser.skipWhitespace();
|
||||
return value;
|
||||
}
|
||||
|
||||
private static final class Parser {
|
||||
private final String text;
|
||||
private int pos;
|
||||
|
||||
Parser(String text) {
|
||||
this.text = text;
|
||||
}
|
||||
|
||||
Object parseValue() {
|
||||
skipWhitespace();
|
||||
char c = text.charAt(pos);
|
||||
if (c == '{') return parseObject();
|
||||
if (c == '[') return parseArray();
|
||||
if (c == '"') return parseString();
|
||||
if (c == 't') { expect("true"); return Boolean.TRUE; }
|
||||
if (c == 'f') { expect("false"); return Boolean.FALSE; }
|
||||
if (c == 'n') { expect("null"); return null; }
|
||||
return parseNumber();
|
||||
}
|
||||
|
||||
Map<String, Object> parseObject() {
|
||||
Map<String, Object> map = new LinkedHashMap<>();
|
||||
pos++; // {
|
||||
skipWhitespace();
|
||||
if (peek() == '}') { pos++; return map; }
|
||||
while (true) {
|
||||
skipWhitespace();
|
||||
String key = parseString();
|
||||
skipWhitespace();
|
||||
if (peek() != ':') throw error("expected ':'");
|
||||
pos++;
|
||||
Object value = parseValue();
|
||||
map.put(key, value);
|
||||
skipWhitespace();
|
||||
char c = text.charAt(pos++);
|
||||
if (c == '}') break;
|
||||
if (c != ',') throw error("expected ',' or '}'");
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
List<Object> parseArray() {
|
||||
List<Object> list = new ArrayList<>();
|
||||
pos++; // [
|
||||
skipWhitespace();
|
||||
if (peek() == ']') { pos++; return list; }
|
||||
while (true) {
|
||||
list.add(parseValue());
|
||||
skipWhitespace();
|
||||
char c = text.charAt(pos++);
|
||||
if (c == ']') break;
|
||||
if (c != ',') throw error("expected ',' or ']'");
|
||||
}
|
||||
return list;
|
||||
}
|
||||
|
||||
String parseString() {
|
||||
if (text.charAt(pos) != '"') throw error("expected string");
|
||||
pos++;
|
||||
StringBuilder sb = new StringBuilder();
|
||||
while (true) {
|
||||
char c = text.charAt(pos++);
|
||||
if (c == '"') break;
|
||||
if (c == '\\') {
|
||||
char esc = text.charAt(pos++);
|
||||
switch (esc) {
|
||||
case '"': sb.append('"'); break;
|
||||
case '\\': sb.append('\\'); break;
|
||||
case '/': sb.append('/'); break;
|
||||
case 'n': sb.append('\n'); break;
|
||||
case 'r': sb.append('\r'); break;
|
||||
case 't': sb.append('\t'); break;
|
||||
case 'b': sb.append('\b'); break;
|
||||
case 'f': sb.append('\f'); break;
|
||||
case 'u':
|
||||
String hex = text.substring(pos, pos + 4);
|
||||
sb.append((char) Integer.parseInt(hex, 16));
|
||||
pos += 4;
|
||||
break;
|
||||
default: throw error("bad escape");
|
||||
}
|
||||
} else {
|
||||
sb.append(c);
|
||||
}
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
Double parseNumber() {
|
||||
int start = pos;
|
||||
while (pos < text.length() && "-+.eE0123456789".indexOf(text.charAt(pos)) >= 0) pos++;
|
||||
return Double.parseDouble(text.substring(start, pos));
|
||||
}
|
||||
|
||||
void expect(String literal) {
|
||||
if (!text.startsWith(literal, pos)) throw error("expected '" + literal + "'");
|
||||
pos += literal.length();
|
||||
}
|
||||
|
||||
char peek() {
|
||||
return text.charAt(pos);
|
||||
}
|
||||
|
||||
void skipWhitespace() {
|
||||
while (pos < text.length() && Character.isWhitespace(text.charAt(pos))) pos++;
|
||||
}
|
||||
|
||||
RuntimeException error(String message) {
|
||||
return new IllegalArgumentException("JSON parse error at " + pos + ": " + message);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,24 @@
|
||||
package com.octoturge.mcmapper.common.protocol;
|
||||
|
||||
/**
|
||||
* A single block change, produced by either the event-driven capture path or the periodic
|
||||
* reconciliation sweep. {@code blockStateId} is a palette index into whatever block/state
|
||||
* palette the leaf module's {@code ChunkAdapter} maintains — the common module never
|
||||
* interprets it, it just carries it to the backend.
|
||||
* For Phase 1 (column/heightmap chunk store), this represents the *resulting* top-of-column
|
||||
* state after a change, not a raw per-voxel diff: {@code y} is the height of the topmost
|
||||
* non-air block at {@code (x, z)}, and {@code blockStateId} is that block's encoded id — the
|
||||
* mod recomputes this itself (it has full world access) rather than the backend trying to infer
|
||||
* a post-break top block from a bare coordinate. Full per-voxel deltas (needed for Phase 2's 3D
|
||||
* meshing) are a natural extension of this same type once the chunk store grows a full
|
||||
* block-data column — see MCMapper-Backend's {@code chunk_columns} table.
|
||||
*
|
||||
* {@code blockStateId} is a palette index into whatever block/state palette the leaf module's
|
||||
* {@code ChunkAdapter} maintains — the common module never interprets it, it just carries it to
|
||||
* the backend. Pre-Flattening leaves (1.7.10/1.12.2) use the identity encoding
|
||||
* {@code (blockId << 4) | meta} (numeric block id/meta *is* the palette); a dynamic palette is
|
||||
* only needed once a leaf's block identity doesn't fit 12+4 bits (Phase 10's NeoForge leaf).
|
||||
*
|
||||
* {@code dimension} is a string so NeoForge's namespaced dimension keys (Phase 10) fit without
|
||||
* changing this type; pre-Flattening leaves always populate it with a stringified vanilla
|
||||
* dimension id (e.g. {@code "0"}), which is what lets {@code DefaultBackendConnection}
|
||||
* {@code Integer.parseInt} it for the wire protocol's integer {@code dimension} field.
|
||||
*/
|
||||
public class DeltaEvent {
|
||||
public final String dimension;
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
package com.octoturge.mcmapper.common.ws;
|
||||
|
||||
import java.io.DataInputStream;
|
||||
import java.io.DataOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.net.Socket;
|
||||
import java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.SecureRandom;
|
||||
import java.util.Base64;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* A minimal RFC 6455 WebSocket client: handshake + masked text-frame send + unmasked text-frame
|
||||
* receive (with fragment reassembly) + ping/pong. Hand-rolled instead of pulling in a library
|
||||
* (e.g. Java-WebSocket) to avoid shading a third-party dependency through legacy ForgeGradle's
|
||||
* classpath — see DefaultBackendConnection's javadoc. No TLS support (`wss://`) — MCMapper's
|
||||
* backend traffic is expected to stay on a private/docker network or behind a TLS-terminating
|
||||
* reverse proxy the mod doesn't need to speak TLS to directly; add {@code SSLSocket} here if a
|
||||
* future deployment needs the mod to dial a public `wss://` endpoint directly.
|
||||
*/
|
||||
public class SimpleWebSocketClient {
|
||||
public interface Listener {
|
||||
void onOpen();
|
||||
|
||||
void onText(String message);
|
||||
|
||||
void onClose(int code, String reason);
|
||||
|
||||
void onError(Exception e);
|
||||
}
|
||||
|
||||
private final Listener listener;
|
||||
private Socket socket;
|
||||
private DataOutputStream out;
|
||||
private Thread readerThread;
|
||||
private final AtomicBoolean open = new AtomicBoolean(false);
|
||||
private final Object writeLock = new Object();
|
||||
|
||||
public SimpleWebSocketClient(Listener listener) {
|
||||
this.listener = listener;
|
||||
}
|
||||
|
||||
public boolean isOpen() {
|
||||
return open.get();
|
||||
}
|
||||
|
||||
public void connect(URI uri) throws IOException {
|
||||
if ("wss".equalsIgnoreCase(uri.getScheme())) {
|
||||
throw new IOException("wss:// is not supported by this client — use ws:// (see class javadoc)");
|
||||
}
|
||||
int port = uri.getPort() == -1 ? 80 : uri.getPort();
|
||||
socket = new Socket(uri.getHost(), port);
|
||||
out = new DataOutputStream(socket.getOutputStream());
|
||||
|
||||
String key = Base64.getEncoder().encodeToString(randomBytes(16));
|
||||
String path = uri.getRawPath() == null || uri.getRawPath().isEmpty() ? "/" : uri.getRawPath();
|
||||
if (uri.getRawQuery() != null) path += "?" + uri.getRawQuery();
|
||||
|
||||
String request = "GET " + path + " HTTP/1.1\r\n"
|
||||
+ "Host: " + uri.getHost() + ":" + port + "\r\n"
|
||||
+ "Upgrade: websocket\r\n"
|
||||
+ "Connection: Upgrade\r\n"
|
||||
+ "Sec-WebSocket-Key: " + key + "\r\n"
|
||||
+ "Sec-WebSocket-Version: 13\r\n"
|
||||
+ "\r\n";
|
||||
out.write(request.getBytes(StandardCharsets.US_ASCII));
|
||||
out.flush();
|
||||
|
||||
InputStream in = socket.getInputStream();
|
||||
String statusLine = readLine(in);
|
||||
if (statusLine == null || !statusLine.contains("101")) {
|
||||
throw new IOException("WebSocket handshake failed, status line: " + statusLine);
|
||||
}
|
||||
String line;
|
||||
while ((line = readLine(in)) != null && !line.isEmpty()) {
|
||||
// Headers ignored beyond the status line — this client trusts the server rather than
|
||||
// verifying Sec-WebSocket-Accept, acceptable for a private/trusted backend endpoint.
|
||||
}
|
||||
|
||||
open.set(true);
|
||||
listener.onOpen();
|
||||
|
||||
readerThread = new Thread(this::readLoop, "mcmapper-ws-reader");
|
||||
readerThread.setDaemon(true);
|
||||
readerThread.start();
|
||||
}
|
||||
|
||||
public void sendText(String text) throws IOException {
|
||||
byte[] payload = text.getBytes(StandardCharsets.UTF_8);
|
||||
synchronized (writeLock) {
|
||||
writeFrame(0x1, payload);
|
||||
}
|
||||
}
|
||||
|
||||
public void close() {
|
||||
if (!open.compareAndSet(true, false)) return;
|
||||
try {
|
||||
synchronized (writeLock) {
|
||||
writeFrame(0x8, new byte[0]);
|
||||
}
|
||||
} catch (IOException ignored) {
|
||||
}
|
||||
closeSocketQuietly();
|
||||
}
|
||||
|
||||
private void closeSocketQuietly() {
|
||||
try {
|
||||
if (socket != null) socket.close();
|
||||
} catch (IOException ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
private void writeFrame(int opcode, byte[] payload) throws IOException {
|
||||
out.writeByte(0x80 | opcode); // FIN + opcode, no extensions
|
||||
|
||||
byte[] mask = randomBytes(4);
|
||||
int len = payload.length;
|
||||
if (len <= 125) {
|
||||
out.writeByte(0x80 | len);
|
||||
} else if (len <= 0xFFFF) {
|
||||
out.writeByte(0x80 | 126);
|
||||
out.writeShort(len);
|
||||
} else {
|
||||
out.writeByte(0x80 | 127);
|
||||
out.writeLong(len);
|
||||
}
|
||||
out.write(mask);
|
||||
byte[] masked = new byte[payload.length];
|
||||
for (int i = 0; i < payload.length; i++) {
|
||||
masked[i] = (byte) (payload[i] ^ mask[i % 4]);
|
||||
}
|
||||
out.write(masked);
|
||||
out.flush();
|
||||
}
|
||||
|
||||
private void readLoop() {
|
||||
try {
|
||||
DataInputStream in = new DataInputStream(socket.getInputStream());
|
||||
StringBuilder fragmentBuffer = new StringBuilder();
|
||||
boolean fragmenting = false;
|
||||
|
||||
while (open.get()) {
|
||||
int b0 = in.readUnsignedByte();
|
||||
boolean fin = (b0 & 0x80) != 0;
|
||||
int opcode = b0 & 0x0F;
|
||||
|
||||
int b1 = in.readUnsignedByte();
|
||||
boolean masked = (b1 & 0x80) != 0; // servers must not mask, but tolerate it
|
||||
long len = b1 & 0x7F;
|
||||
if (len == 126) {
|
||||
len = in.readUnsignedShort();
|
||||
} else if (len == 127) {
|
||||
len = in.readLong();
|
||||
}
|
||||
byte[] serverMask = masked ? readFully(in, 4) : null;
|
||||
byte[] payload = readFully(in, (int) len);
|
||||
if (serverMask != null) {
|
||||
for (int i = 0; i < payload.length; i++) payload[i] ^= serverMask[i % 4];
|
||||
}
|
||||
|
||||
switch (opcode) {
|
||||
case 0x1: // text
|
||||
case 0x0: // continuation
|
||||
fragmenting = true;
|
||||
fragmentBuffer.append(new String(payload, StandardCharsets.UTF_8));
|
||||
if (fin) {
|
||||
fragmenting = false;
|
||||
String message = fragmentBuffer.toString();
|
||||
fragmentBuffer.setLength(0);
|
||||
listener.onText(message);
|
||||
}
|
||||
break;
|
||||
case 0x8: // close
|
||||
open.set(false);
|
||||
int code = payload.length >= 2 ? ((payload[0] & 0xFF) << 8 | (payload[1] & 0xFF)) : 1000;
|
||||
String reason = payload.length > 2
|
||||
? new String(payload, 2, payload.length - 2, StandardCharsets.UTF_8)
|
||||
: "";
|
||||
listener.onClose(code, reason);
|
||||
closeSocketQuietly();
|
||||
return;
|
||||
case 0x9: // ping
|
||||
synchronized (writeLock) {
|
||||
writeFrame(0xA, payload);
|
||||
}
|
||||
break;
|
||||
case 0xA: // pong
|
||||
break;
|
||||
default:
|
||||
// unknown opcode — ignore rather than fail the connection
|
||||
}
|
||||
if (!fin && !fragmenting) {
|
||||
// defensive: a non-fin non-text/continuation frame we don't understand
|
||||
}
|
||||
}
|
||||
} catch (IOException e) {
|
||||
if (open.compareAndSet(true, false)) {
|
||||
listener.onError(e);
|
||||
listener.onClose(1006, "connection lost: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] readFully(DataInputStream in, int len) throws IOException {
|
||||
byte[] buf = new byte[len];
|
||||
in.readFully(buf);
|
||||
return buf;
|
||||
}
|
||||
|
||||
private static String readLine(InputStream in) throws IOException {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
int c;
|
||||
boolean any = false;
|
||||
while ((c = in.read()) != -1) {
|
||||
any = true;
|
||||
if (c == '\r') continue;
|
||||
if (c == '\n') break;
|
||||
sb.append((char) c);
|
||||
}
|
||||
return any ? sb.toString() : null;
|
||||
}
|
||||
|
||||
private static final SecureRandom RANDOM = new SecureRandom();
|
||||
|
||||
private static byte[] randomBytes(int n) {
|
||||
byte[] b = new byte[n];
|
||||
RANDOM.nextBytes(b);
|
||||
return b;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user