From 56f13ac2334dc414b04016172e8d04c0f47d321d Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:39:43 +0200 Subject: [PATCH 1/7] Stabilize NetherNet channels for long running servers Collected fixes from running the transport on production servers: unique placeholder remote addresses per connection instead of a shared 0.0.0.0 (keeps per IP limits and throttling functional), a guarded single fire channelActive, signaling stability fixes for long uptimes, a WebSocket frame aggregator on the signaling pipeline, reassembly buffer release on close, an outbound fragmentation bound, and notes on the reliable only channel design. --- .../channel/nethernet/NetherNetChannel.java | 165 +++++++++++++----- .../nethernet/NetherNetClientChannel.java | 7 +- .../channel/nethernet/NetherNetConstants.java | 30 ++++ .../nethernet/NetherNetServerChannel.java | 87 ++++++++- .../AbstractNetherNetXboxSignaling.java | 45 +++++ .../signaling/NetherNetXboxSignaling.java | 23 ++- 6 files changed, 306 insertions(+), 51 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java index 3318d4d..c1ef4d2 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java @@ -22,6 +22,7 @@ import java.nio.ByteBuffer; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; public abstract class NetherNetChannel extends AbstractChannel { private static final InternalLogger log = InternalLoggerFactory.getInstance(NetherNetChannel.class); @@ -36,9 +37,27 @@ public abstract class NetherNetChannel extends AbstractChannel { protected RTCDataChannel unreliableChannel; protected final Queue pendingWrites = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean channelActiveFired = new AtomicBoolean(); protected volatile boolean open = true; + // Reassembly state for inbound fragmented messages. assemblyBuf is written on + // the WebRTC callback thread (the data channel observer) and released on the + // event loop (doClose), so every access is guarded by assemblyLock together + // with the assemblyClosed flag, which orders the release strictly after the + // observer is detached and prevents a use-after-free. + private final Object assemblyLock = new Object(); + private ByteBuf assemblyBuf; + private boolean assemblyClosed; + private int currentSegmentCount = -1; + + // Maximum outbound SCTP message size in bytes. writeInternal must never + // fragment beyond what the remote advertised it can receive (the + // a=max-message-size attribute in its SDP); subclasses set this from the + // negotiated description. Defaults to the conservative MAX_SCTP_MESSAGE_SIZE + // until negotiation supplies the real value. + protected volatile int maxOutboundMessageSize = NetherNetConstants.MAX_SCTP_MESSAGE_SIZE; + protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketAddress local) { super(parent); this.remoteAddress = remote; @@ -47,12 +66,23 @@ protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketA public void setDataChannels(RTCDataChannel reliable, RTCDataChannel unreliable) { this.reliableChannel = reliable; + // The unreliable channel is intentionally stored but never observed or + // written to: this is a reliable-only implementation, which gives the + // ordered, lossless stream a Bedrock translation proxy needs. If it is + // ever wired up it must stay single-message only, because an unreliable + // channel can drop or reorder and would strand a multi-fragment message + // mid-reassembly: observe it but never run reassembly on it (every + // inbound message must arrive complete, header 0), and on send drop + // anything too large to fit one message rather than fragmenting it. this.unreliableChannel = unreliable; - RTCDataChannelObserver observer = new RTCDataChannelObserver() { - private final ByteBuf assemblyBuf = config.getAllocator().buffer(); - private int currentSegmentCount = -1; + synchronized (assemblyLock) { + if (assemblyBuf == null && !assemblyClosed) { + assemblyBuf = config.getAllocator().buffer(); + } + } + RTCDataChannelObserver observer = new RTCDataChannelObserver() { @Override public void onBufferedAmountChange(long previousAmount) { } @@ -70,39 +100,45 @@ public void onMessage(RTCDataChannelBuffer buffer) { int segments = data.get() & 0xFF; - if (currentSegmentCount == -1) { - currentSegmentCount = segments; - } else { - if (segments != currentSegmentCount - 1) { - assemblyBuf.clear(); - currentSegmentCount = -1; + synchronized (assemblyLock) { + if (assemblyClosed || assemblyBuf == null) { return; } - currentSegmentCount = segments; - } - if (data.hasRemaining()) { - byte[] payload = new byte[data.remaining()]; - data.get(payload); - assemblyBuf.writeBytes(payload); - } + if (currentSegmentCount == -1) { + currentSegmentCount = segments; + } else { + if (segments != currentSegmentCount - 1) { + assemblyBuf.clear(); + currentSegmentCount = -1; + return; + } + currentSegmentCount = segments; + } - if (segments == 0) { - try { - if (assemblyBuf.isReadable()) { - ByteBuf packet = assemblyBuf.copy(); - assemblyBuf.skipBytes(assemblyBuf.readableBytes()); + if (data.hasRemaining()) { + byte[] payload = new byte[data.remaining()]; + data.get(payload); + assemblyBuf.writeBytes(payload); + } - eventLoop().execute(() -> { - pipeline().fireChannelRead(packet); - pipeline().fireChannelReadComplete(); - }); + if (segments == 0) { + try { + if (assemblyBuf.isReadable()) { + ByteBuf packet = assemblyBuf.copy(); + + eventLoop().execute(() -> { + fireChannelActiveIfReady(); + pipeline().fireChannelRead(packet); + pipeline().fireChannelReadComplete(); + }); + } + } catch (Exception e) { + log.error("Error processing packet", e); + } finally { + assemblyBuf.clear(); + currentSegmentCount = -1; } - } catch (Exception e) { - log.error("Error processing packet", e); - } finally { - assemblyBuf.clear(); - currentSegmentCount = -1; } } } @@ -115,17 +151,42 @@ public void onMessage(RTCDataChannelBuffer buffer) { } } + /** + * Sets the maximum outbound SCTP message size, in bytes, for this channel. + * Should be the {@code a=max-message-size} the remote peer advertised in its + * SDP. Values too small to leave room for the fragment header are ignored. + * + * @param size the negotiated maximum message size + */ + public void setMaxOutboundMessageSize(int size) { + if (size > 1) { + this.maxOutboundMessageSize = size; + } + } + private void onDataChannelStateChange() { if (isActive()) { - if (!pendingWrites.isEmpty()) { - pipeline().fireChannelWritabilityChanged(); - unsafe().flush(); - } + fireChannelActiveIfReady(); } else if (reliableChannel.getState() == RTCDataChannelState.CLOSED) { close(); } } + protected void fireChannelActiveIfReady() { + if (!isRegistered() || !isActive()) { + return; + } + + if (channelActiveFired.compareAndSet(false, true)) { + pipeline().fireChannelActive(); + } + + if (!pendingWrites.isEmpty()) { + pipeline().fireChannelWritabilityChanged(); + unsafe().flush(); + } + } + @Override protected void doWrite(ChannelOutboundBuffer in) throws Exception { if (!isActive()) { @@ -160,15 +221,26 @@ private void writeInternal(Object msg) { ByteBuf payload = (ByteBuf) msg; - ByteBuf framed = payload.retainedDuplicate(); - - int totalLength = framed.readableBytes(); - int maxPayload = NetherNetConstants.MAX_SCTP_MESSAGE_SIZE - 1; + int maxPayload = maxOutboundMessageSize - 1; + int totalLength = payload.readableBytes(); int segments = (totalLength / maxPayload); if (totalLength % maxPayload != 0) segments++; + // Each fragment carries a one-byte countdown header, so a message can + // span at most 256 fragments. The negotiated max-message-size keeps this + // ceiling far out of reach, but guard it anyway: without the check an + // oversized message would wrap the header byte and silently corrupt the + // stream instead of failing. + if (segments > 256) { + pipeline().fireExceptionCaught(new IllegalStateException( + "Outbound message of " + totalLength + " bytes exceeds the maximum " + + (256 * maxPayload) + " bytes addressable by the fragment header")); + return; + } + + ByteBuf framed = payload.retainedDuplicate(); try { int offset = 0; for (int i = 0; i < segments; i++) { @@ -194,6 +266,9 @@ private void writeInternal(Object msg) { @Override protected void doRegister() throws Exception { + if (isActive()) { + channelActiveFired.set(true); + } } @Override @@ -226,6 +301,18 @@ protected void doClose() throws Exception { peerConnection.close(); } + // Release the reassembly buffer now that the observer is unregistered. + // Guarded by assemblyLock with assemblyClosed so it cannot race, or be + // re-allocated by, an in-flight onMessage on the WebRTC callback thread. + synchronized (assemblyLock) { + assemblyClosed = true; + if (assemblyBuf != null) { + assemblyBuf.release(); + assemblyBuf = null; + } + currentSegmentCount = -1; + } + Object msg; while ((msg = pendingWrites.poll()) != null) { ReferenceCountUtil.release(msg); @@ -270,4 +357,4 @@ public boolean isActive() { public ChannelMetadata metadata() { return METADATA; } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java index 8749323..fd5e463 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java @@ -296,6 +296,9 @@ private void handleSignal(String signal) { switch (type) { case NetherNetConstants.RTC_NEGOTIATION_CONNECT_RESPONSE -> { + // Fragment outbound data no larger than the remote advertised + // it can receive (a=max-message-size in its answer). + setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(data, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); peerConnection.setRemoteDescription(new RTCSessionDescription(RTCSdpType.ANSWER, data), new SetSessionDescriptionObserver() { @Override public void onSuccess() {} @Override public void onFailure(String e) { /* Retry handled by timeout */ } @@ -348,7 +351,7 @@ public void onStateChange() { if (connectPromise != null && !connectPromise.isDone()) { connectPromise.trySuccess(); } - pipeline().fireChannelActive(); + fireChannelActiveIfReady(); } }); } @@ -364,4 +367,4 @@ private long cycleConnectionId() { this.connectionId = ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE); return this.connectionId; } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java index dcc6a0f..3df9546 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java @@ -177,4 +177,34 @@ public static String buildSignalConnectResponse(long connectionId, String sdp) { public static String buildSignalCandidateAdd(long connectionId, String candidateSdp) { return RTC_NEGOTIATION_CANDIDATE_ADD + " " + Long.toUnsignedString(connectionId) + " " + candidateSdp; } + + /** + * Parses the {@code a=max-message-size} attribute from an SDP description. + * This is the maximum SCTP user message size, in bytes, that the describing + * endpoint is willing to receive, so outbound fragmentation must never + * exceed the value advertised by the remote peer. + * + * @param sdp the SDP description to scan, may be null + * @param fallback the value to return when the attribute is absent or invalid + * @return the advertised maximum message size, or {@code fallback} if not found + */ + public static int parseMaxMessageSize(String sdp, int fallback) { + if (sdp == null) { + return fallback; + } + for (String line : sdp.split("\\r?\\n")) { + String trimmed = line.trim(); + if (trimmed.startsWith("a=max-message-size:")) { + try { + int value = Integer.parseInt(trimmed.substring("a=max-message-size:".length()).trim()); + if (value > 1) { + return value; + } + } catch (NumberFormatException ignored) { + // Fall through to the fallback for a malformed attribute. + } + } + } + return fallback; + } } \ No newline at end of file diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java index 1eef7f7..fc8480e 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java @@ -29,6 +29,7 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.List; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; public class NetherNetServerChannel extends AbstractServerChannel { @@ -96,7 +97,11 @@ public void acceptConnection(long connectionId, String offerSdp, String remoteNe ServerPeerConnectionObserver observer = new ServerPeerConnectionObserver(connectionId, remoteNetworkId); RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, observer); - NetherNetChildChannel child = new NetherNetChildChannel(this, pc, new InetSocketAddress(0), localAddress); + NetherNetChildChannel child = new NetherNetChildChannel(this, pc, generatePlaceholderAddress(), localAddress); + // Fragment outbound data no larger than the client advertised it can + // receive (a=max-message-size in its offer), falling back to the + // conservative default when the client does not advertise one. + child.setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(offerSdp, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); observer.setChildChannel(child); child.closeFuture().addListener(future -> signaling.removeSignalHandler(connectionId)); @@ -172,6 +177,7 @@ private class ServerPeerConnectionObserver implements PeerConnectionObserver { private RTCDataChannel reliable; private RTCDataChannel unreliable; + private boolean dataChannelsSet; private ScheduledFuture handshakeTimeout; @@ -237,19 +243,86 @@ public void onDataChannel(RTCDataChannel dataChannel) { } private void checkDataChannels() { - if (child != null && reliable != null && unreliable != null) { + if (child != null && reliable != null && unreliable != null && !dataChannelsSet) { + dataChannelsSet = true; if (handshakeTimeout != null) { handshakeTimeout.cancel(false); } log.debug("Data Channels established for {}", Long.toUnsignedString(this.connectionId)); child.setDataChannels(reliable, unreliable); - - if (child.pipeline() != null) { - child.pipeline().fireChannelActive(); - } + child.fireChannelActiveIfReady(); } } + + // TODO: Resolve real remote address from ICE candidate pair via webrtc-java fork. + // The code below works but delays fireChannelActive, which causes pipeline timing + // issues (duplicate handler registration). Requires a synchronous getter in the + // native JNI layer to avoid the delay. Until then, connections use a unique + // 10.x.x.x placeholder address set at channel creation time. + // + // private void resolveRemoteAddressThenActivate(NetherNetChildChannel child) { + // java.util.concurrent.atomic.AtomicBoolean activated = new java.util.concurrent.atomic.AtomicBoolean(false); + // ScheduledFuture fallback = eventLoop().schedule(() -> { + // activateOnce(child, activated); + // }, 2, TimeUnit.SECONDS); + // try { + // child.peerConnection.getStats(report -> { + // try { + // String remoteId = null; + // for (var entry : report.getStats().entrySet()) { + // var stats = entry.getValue(); + // if (stats.getType() == dev.kastle.webrtc.RTCStatsType.CANDIDATE_PAIR) { + // Object selected = stats.getAttributes().get("nominated"); + // if (Boolean.TRUE.equals(selected)) { + // remoteId = (String) stats.getAttributes().get("remoteCandidateId"); + // break; + // } + // } + // } + // if (remoteId != null) { + // var remoteStat = report.getStats().get(remoteId); + // if (remoteStat != null) { + // String ip = (String) remoteStat.getAttributes().get("ip"); + // Object portObj = remoteStat.getAttributes().get("port"); + // if (ip != null && portObj != null) { + // int port = (portObj instanceof Number) ? ((Number) portObj).intValue() : Integer.parseInt(portObj.toString()); + // child.remoteAddress = new InetSocketAddress(ip, port); + // } + // } + // } + // } catch (Exception e) { + // log.debug("Failed to resolve remote address: {}", e.getMessage()); + // } finally { + // fallback.cancel(false); + // activateOnce(child, activated); + // } + // }); + // } catch (Exception e) { + // fallback.cancel(false); + // activateOnce(child, activated); + // } + // } + // + // private void activateOnce(NetherNetChildChannel child, java.util.concurrent.atomic.AtomicBoolean activated) { + // if (!activated.compareAndSet(false, true)) return; + // child.eventLoop().execute(() -> { + // if (child.isOpen() && child.pipeline() != null) { + // child.pipeline().fireChannelActive(); + // } + // }); + // } + } + + /** + * Generates a unique placeholder address in the 10.x.x.x range for a new + * Nethernet connection. The 10.0.0.0/8 range is private (RFC 1918) and + * will not collide with real public client addresses. + */ + private static InetSocketAddress generatePlaceholderAddress() { + ThreadLocalRandom r = ThreadLocalRandom.current(); + String ip = "10." + (r.nextInt(1, 256)) + "." + (r.nextInt(256)) + "." + (r.nextInt(1, 256)); + return new InetSocketAddress(ip, 0); } @Override @@ -295,4 +368,4 @@ public boolean isActive() { public ChannelMetadata metadata() { return METADATA; } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java index dde7b09..ccde7df 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java @@ -52,6 +52,7 @@ public abstract class AbstractNetherNetXboxSignaling extends SimpleChannelInboun protected Channel channel; protected CompletableFuture> connectFuture; protected volatile List iceServers = new ArrayList<>(); + protected volatile long lastMessageReceivedAt; protected final Map handlers = new ConcurrentHashMap<>(); protected NetherNetServerSignaling.NewConnectionHandler newConnectionHandler; @@ -132,12 +133,56 @@ protected void initChannel(SocketChannel ch) { public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) { log.debug("{} WebSocket Connected", getClass().getSimpleName()); + lastMessageReceivedAt = System.currentTimeMillis(); onConnected(ctx); } else { super.userEventTriggered(ctx, evt); } } + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + // Track liveness: every inbound frame counts. Used by isChannelAlive() + // to detect silent half-closed TCP where channel.isActive() lies. + lastMessageReceivedAt = System.currentTimeMillis(); + super.channelRead(ctx, msg); + } + + /** + * @return true if the signaling WebSocket channel is active. Does NOT + * detect silent half-closed TCP (Netty can report active on a + * dead socket). For stricter checking use {@link #isChannelAlive(long)} + * or compare {@link #getMillisSinceLastMessage()} against your + * own threshold. + */ + public boolean isChannelAlive() { + Channel c = this.channel; + return c != null && c.isActive(); + } + + /** + * @param maxSilenceMillis max tolerated time since last received frame. + * Note: a server that's not receiving any client + * connections may legitimately go long periods + * without inbound messages. + * @return true if the channel is active and has received a frame within + * the given window. + */ + public boolean isChannelAlive(long maxSilenceMillis) { + if (!isChannelAlive()) return false; + long silence = getMillisSinceLastMessage(); + return silence >= 0 && silence <= maxSilenceMillis; + } + + /** + * @return milliseconds since the last received frame, or -1 if no frame + * has been received yet. + */ + public long getMillisSinceLastMessage() { + if (lastMessageReceivedAt == 0) return -1; + return System.currentTimeMillis() - lastMessageReceivedAt; + } + /** * Called when the WebSocket handshake is complete. */ diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java index 8e00d65..0bd0072 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java @@ -47,10 +47,27 @@ public NetherNetXboxSignaling(String xboxToken) { @Override protected void onConnected(ChannelHandlerContext ctx) { + // IMPORTANT: scheduleAtFixedRate silently cancels the task on the first + // exception thrown by the runnable. A single transient write failure would + // kill the ping loop forever, leaving the connection to die to idle timeouts. + // We wrap in try/catch and add a write listener so failures are logged + // but the task keeps running on its regular cadence. ctx.executor().scheduleAtFixedRate(() -> { - JsonObject ping = new JsonObject(); - ping.addProperty("Type", 0); - ctx.writeAndFlush(new TextWebSocketFrame(gson.toJson(ping))); + try { + if (!ctx.channel().isActive()) { + return; + } + JsonObject ping = new JsonObject(); + ping.addProperty("Type", 0); + ctx.writeAndFlush(new TextWebSocketFrame(gson.toJson(ping))) + .addListener(future -> { + if (!future.isSuccess()) { + log.warn("Ping write failed: {}", future.cause() != null ? future.cause().getMessage() : "unknown"); + } + }); + } catch (Throwable t) { + log.warn("Ping iteration threw; loop continues: {}", t.getMessage()); + } }, 5, 5, TimeUnit.SECONDS); } From a936e437ff06e9491de13d2ede11be35e36e310e Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:21:22 +0200 Subject: [PATCH 2/7] Make signaling WebSockets reconnectable in place reconnect(freshToken) replaces only the socket to the signaling service. The signaling instance, its handlers, and everything built on it (server channel, WebRTC factories, live peer connections) survive, so a signaling drop no longer requires tearing the transport down. Liveness is now detectable on idle servers: a WebSocket protocol ping every 15 seconds guarantees inbound pongs on a healthy socket, so isChannelAlive(maxSilence) also catches silently half open TCP. Per channel scheduled tasks are tracked and cancelled on channel inactive, so reconnects no longer leak ping loops. TURN credential pushes are applied for the lifetime of the socket instead of only during connect, and the JSON RPC endpoint refreshes credentials every 30 minutes, so late joining peers no longer receive expired relay credentials. Pending RPC requests fail fast when the socket dies. The frame aggregator limit is raised to 128 KB for batched RPC frames. --- .../AbstractNetherNetXboxSignaling.java | 261 +++++++++++++++--- .../signaling/NetherNetXboxRpcSignaling.java | 66 +++-- .../signaling/NetherNetXboxSignaling.java | 52 ++-- 3 files changed, 289 insertions(+), 90 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java index ccde7df..6e46e08 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java @@ -8,6 +8,7 @@ import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; @@ -17,6 +18,7 @@ import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory; @@ -25,6 +27,8 @@ import io.netty.handler.codec.http.websocketx.WebSocketVersion; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; +import io.netty.util.AttributeKey; +import io.netty.util.concurrent.ScheduledFuture; import io.netty.util.internal.logging.InternalLogger; import io.netty.util.internal.logging.InternalLoggerFactory; @@ -38,21 +42,37 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; -public abstract class AbstractNetherNetXboxSignaling extends SimpleChannelInboundHandler +public abstract class AbstractNetherNetXboxSignaling extends SimpleChannelInboundHandler implements NetherNetClientSignaling, NetherNetServerSignaling { - + + /** Time to wait for the connect plus credential exchange before giving up. */ + private static final long CONNECT_TIMEOUT_SECONDS = 20; + /** Interval for WebSocket protocol-level pings. The RFC obliges the server to + * answer with a pong, so these guarantee inbound traffic on a healthy socket + * and make {@link #isChannelAlive(long)} reliable even on idle servers. */ + private static final long WS_PING_INTERVAL_SECONDS = 15; + + /** Recurring tasks tied to one WebSocket channel's lifetime, cancelled in + * channelInactive. Without this, every reconnect would leak the previous + * channel's ping loops on the shared event loop. */ + private static final AttributeKey>> CHANNEL_TASKS = + AttributeKey.valueOf("nethernet-signaling-channel-tasks"); + protected final InternalLogger log = InternalLoggerFactory.getInstance(getClass()); - protected final String xboxToken; + protected volatile String xboxToken; protected final String localNetworkId; protected final URI uri; protected final EventLoopGroup eventLoopGroup; - - protected Channel channel; + + protected volatile Channel channel; protected CompletableFuture> connectFuture; protected volatile List iceServers = new ArrayList<>(); protected volatile long lastMessageReceivedAt; + private volatile boolean closed; protected final Map handlers = new ConcurrentHashMap<>(); protected NetherNetServerSignaling.NewConnectionHandler newConnectionHandler; @@ -77,28 +97,60 @@ public synchronized CompletableFuture> connect(SocketAddress @Override public void bind(SocketAddress localAddress) throws ConnectException { - try { - connectInternal().join(); - } catch (Exception e) { - Throwable cause = e.getCause() != null ? e.getCause() : e; - close(); - if (cause instanceof ConnectException) throw (ConnectException) cause; - ConnectException ce = new ConnectException("Failed to connect to Xbox Signaling: " + cause.getMessage()); - ce.initCause(cause); - throw ce; + joinConnect(connectInternal()); + } + + /** + * Reconnects the WebSocket in place using a fresh authorization token. The + * signaling instance, its registered handlers, and everything built on top + * of it (server channel, peer connection factory, live peer connections) + * are untouched; only the socket to the signaling service is replaced. + * + * Must not be called from this signaling instance's own event loop; it + * blocks until the new socket has completed its credential exchange. + * + * @param freshToken the authorization token to connect with + * @throws ConnectException if the reconnect fails; the instance remains + * reconnectable and the call may be retried + */ + public void reconnect(String freshToken) throws ConnectException { + CompletableFuture> future; + synchronized (this) { + if (closed) { + throw new ConnectException("Signaling has been closed"); + } + this.xboxToken = freshToken; + Channel old = this.channel; + this.channel = null; + CompletableFuture> pending = this.connectFuture; + this.connectFuture = null; + if (pending != null && !pending.isDone()) { + pending.completeExceptionally(new ClosedChannelException()); + } + if (old != null) { + old.close(); + } + future = connectInternal(); } + joinConnect(future); } protected synchronized CompletableFuture> connectInternal() { + if (closed) { + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new ClosedChannelException()); + return failed; + } if (connectFuture != null) return connectFuture; - connectFuture = new CompletableFuture<>(); - connectFuture.thenAccept(servers -> this.iceServers = servers); - + CompletableFuture> future = new CompletableFuture<>(); + connectFuture = future; + future.thenAccept(servers -> this.iceServers = servers); + try { SslContext sslCtx = SslContextBuilder.forClient().build(); WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker( - uri, WebSocketVersion.V13, null, false, + uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders() .add("Authorization", xboxToken) .add("User-Agent", NetherNetConstants.SIGNALING_USER_AGENT) @@ -115,18 +167,61 @@ protected void initChannel(SocketChannel ch) { ChannelPipeline p = ch.pipeline(); p.addLast(sslCtx.newHandler(ch.alloc(), uri.getHost(), 443)); p.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192)); - p.addLast("ws-handshake", new WebSocketClientProtocolHandler(handshaker)); - p.addLast("ws-aggregator", new WebSocketFrameAggregator(16 * 1024)); // Allow 16KB aggregations + // dropPongFrames=false: pongs must reach channelRead so they + // refresh lastMessageReceivedAt for liveness detection. + p.addLast("ws-handshake", new WebSocketClientProtocolHandler(handshaker, true, false)); + p.addLast("ws-aggregator", new WebSocketFrameAggregator(128 * 1024)); p.addLast("handler", AbstractNetherNetXboxSignaling.this); } }); - this.channel = b.connect(uri.getHost(), 443).sync().channel(); + // Asynchronous on purpose: connectInternal may be invoked while + // holding this signaling's lock or (via channelInactive paths) on + // the event loop itself, where a sync() would deadlock. Callers + // that need to block use joinConnect on the returned future. + ChannelFuture connect = b.connect(uri.getHost(), 443); + this.channel = connect.channel(); + connect.addListener(f -> { + if (!f.isSuccess() && !future.isDone()) { + future.completeExceptionally(f.cause()); + } + }); } catch (Exception e) { Throwable cause = e.getCause() != null ? e.getCause() : e; - if (connectFuture != null) connectFuture.completeExceptionally(cause); + if (!future.isDone()) future.completeExceptionally(cause); + } + return future; + } + + /** + * Blocks until the given connect attempt has completed its credential + * exchange, translating failures (including a stalled handshake) into a + * ConnectException after cleaning up the half-open channel. + */ + protected void joinConnect(CompletableFuture> future) throws ConnectException { + try { + future.orTimeout(CONNECT_TIMEOUT_SECONDS, TimeUnit.SECONDS).join(); + } catch (Exception e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + abortConnect(); + if (cause instanceof ConnectException) throw (ConnectException) cause; + ConnectException ce = new ConnectException("Failed to connect to Xbox Signaling: " + cause.getMessage()); + ce.initCause(cause); + throw ce; + } + } + + private synchronized void abortConnect() { + Channel c = this.channel; + this.channel = null; + CompletableFuture> pending = this.connectFuture; + this.connectFuture = null; + if (pending != null && !pending.isDone()) { + pending.completeExceptionally(new ClosedChannelException()); + } + if (c != null) { + c.close(); } - return connectFuture; } @Override @@ -134,6 +229,8 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) { log.debug("{} WebSocket Connected", getClass().getSimpleName()); lastMessageReceivedAt = System.currentTimeMillis(); + scheduleRecurring(ctx, "ws-ping", () -> ctx.writeAndFlush(new PingWebSocketFrame()), + WS_PING_INTERVAL_SECONDS, WS_PING_INTERVAL_SECONDS); onConnected(ctx); } else { super.userEventTriggered(ctx, evt); @@ -142,12 +239,40 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - // Track liveness: every inbound frame counts. Used by isChannelAlive() + // Track liveness: every inbound frame counts, including pong frames + // answering our protocol-level pings. Used by isChannelAlive(long) // to detect silent half-closed TCP where channel.isActive() lies. lastMessageReceivedAt = System.currentTimeMillis(); super.channelRead(ctx, msg); } + /** + * Schedules a recurring task bound to the given channel's lifetime. The + * task is skipped while the channel is inactive, survives its own + * exceptions (scheduleAtFixedRate would otherwise cancel silently on the + * first throw), and is cancelled when the channel goes inactive. + */ + protected void scheduleRecurring(ChannelHandlerContext ctx, String name, Runnable task, + long initialDelaySeconds, long periodSeconds) { + ScheduledFuture future = ctx.executor().scheduleAtFixedRate(() -> { + try { + if (!ctx.channel().isActive()) { + return; + } + task.run(); + } catch (Throwable t) { + log.warn("{} task threw; loop continues: {}", name, t.getMessage()); + } + }, initialDelaySeconds, periodSeconds, TimeUnit.SECONDS); + + CopyOnWriteArrayList> tasks = ctx.channel().attr(CHANNEL_TASKS).get(); + if (tasks == null) { + ctx.channel().attr(CHANNEL_TASKS).setIfAbsent(new CopyOnWriteArrayList<>()); + tasks = ctx.channel().attr(CHANNEL_TASKS).get(); + } + tasks.add(future); + } + /** * @return true if the signaling WebSocket channel is active. Does NOT * detect silent half-closed TCP (Netty can report active on a @@ -162,9 +287,11 @@ public boolean isChannelAlive() { /** * @param maxSilenceMillis max tolerated time since last received frame. - * Note: a server that's not receiving any client - * connections may legitimately go long periods - * without inbound messages. + * The recurring protocol-level ping guarantees a + * pong at least every {@value #WS_PING_INTERVAL_SECONDS} + * seconds on a healthy socket, so thresholds of + * two to three times that are safe even when no + * signaling traffic is flowing. * @return true if the channel is active and has received a frame within * the given window. */ @@ -184,10 +311,18 @@ public long getMillisSinceLastMessage() { } /** - * Called when the WebSocket handshake is complete. + * Called when the WebSocket handshake is complete. */ protected abstract void onConnected(ChannelHandlerContext ctx); + /** + * Called when a WebSocket channel of this signaling instance goes + * inactive, before the base class finishes its own cleanup. Subclasses + * use this to fail state tied to the dead socket (e.g. pending requests). + */ + protected void onChannelInactive(ChannelHandlerContext ctx) { + } + @Override public List getIceServers() { return this.iceServers; @@ -212,7 +347,7 @@ public void setSignalHandler(long connectionId, SignalHandler handler) { public void removeSignalHandler(long connectionId) { this.handlers.remove(connectionId); } - + @Override public void setAdvertisementData(PongData pongData) { // No-op for Xbox Signaling. @@ -220,8 +355,10 @@ public void setAdvertisementData(PongData pongData) { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { - if (connectFuture != null && !connectFuture.isDone()) { - connectFuture.completeExceptionally(cause); + synchronized (this) { + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.completeExceptionally(cause); + } } log.error("Signaling Exception: {}", cause.getMessage(), cause); ctx.close(); @@ -229,19 +366,45 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + CopyOnWriteArrayList> tasks = ctx.channel().attr(CHANNEL_TASKS).get(); + if (tasks != null) { + for (ScheduledFuture task : tasks) { + task.cancel(false); + } + tasks.clear(); + } synchronized (this) { - if (connectFuture != null && !connectFuture.isDone()) { - connectFuture.completeExceptionally(new ClosedChannelException()); + // Only clear state if the channel going inactive is still the + // current one. During a reconnect the old channel's inactive event + // arrives after the replacement is already installed and must not + // tear down the new connection's state. + if (ctx.channel() == this.channel) { + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.completeExceptionally(new ClosedChannelException()); + } + connectFuture = null; + this.channel = null; } - connectFuture = null; - this.channel = null; } + onChannelInactive(ctx); super.channelInactive(ctx); } @Override public void close() { - if (channel != null) channel.close(); + Channel c; + synchronized (this) { + closed = true; + c = this.channel; + this.channel = null; + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.completeExceptionally(new ClosedChannelException()); + } + connectFuture = null; + } + if (c != null) { + c.close(); + } eventLoopGroup.shutdownGracefully(); } @@ -252,7 +415,7 @@ protected void dispatchSignalToPipeline(String sender, String rawMsg) { if (parts.length < 2) return; long connectionId = Long.parseUnsignedLong(parts[1]); - + SignalHandler handler = handlers.get(connectionId); if (handler != null) { handler.onSignal(rawMsg); @@ -281,19 +444,19 @@ protected List parseTurnServers(JsonObject json) { for (JsonElement el : servers) { JsonObject server = el.getAsJsonObject(); List urls = new ArrayList<>(); - + JsonArray urlsArray = null; if (server.has("Urls")) urlsArray = server.getAsJsonArray("Urls"); else if (server.has("urls")) urlsArray = server.getAsJsonArray("urls"); if (urlsArray != null) { urlsArray.forEach(u -> urls.add(u.getAsString())); - + IceServerInfo.Builder info = new IceServerInfo.Builder().setUrls(urls); - + if (server.has("Username")) info.setUsername(server.get("Username").getAsString()); else if (server.has("username")) info.setUsername(server.get("username").getAsString()); - + if (server.has("Password")) info.setPassword(server.get("Password").getAsString()); else if (server.has("password")) info.setPassword(server.get("password").getAsString()); else if (server.has("Credential")) info.setPassword(server.get("Credential").getAsString()); @@ -309,4 +472,20 @@ protected List parseTurnServers(JsonObject json) { log.debug("Successfully parsed {} ICE servers.", result.size()); return result; } + + /** + * Stores freshly received TURN/ICE servers and completes the pending + * connect future if this was the credential exchange of a new socket. + * Called by subclasses whenever the service supplies credentials, both + * during connect and on later refreshes, so peer connections created at + * any point get unexpired TURN credentials. + */ + protected void updateIceServers(List servers) { + this.iceServers = servers; + synchronized (this) { + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.complete(servers); + } + } + } } diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java index 4345a77..21ce617 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java @@ -7,26 +7,31 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import dev.kastle.netty.channel.nethernet.NetherNetConstants; +import io.netty.channel.ChannelHandler.Sharable; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import java.net.URI; import java.nio.channels.ClosedChannelException; -import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; +@Sharable public class NetherNetXboxRpcSignaling extends AbstractNetherNetXboxSignaling { private static final Gson gson = new GsonBuilder().serializeNulls().create(); + + /** Interval for proactively re-fetching TURN credentials so peer + * connections created on a long-lived socket never receive expired ones. */ + private static final long TURN_REFRESH_INTERVAL_SECONDS = 30 * 60; + private final Map> pendingRequests = new ConcurrentHashMap<>(); /** * Creates a NetherNetXboxRpcSignaling instance. - * + * * @param networkId The Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -36,7 +41,7 @@ public NetherNetXboxRpcSignaling(String networkId, String xboxToken) { /** * Creates a NetherNetXboxRpcSignaling instance. - * + * * @param localNetworkId The local Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -46,7 +51,7 @@ public NetherNetXboxRpcSignaling(long localNetworkId, String xboxToken) { /** * Creates a NetherNetXboxRpcSignaling instance with a random local Network ID. - * + * * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ public NetherNetXboxRpcSignaling(String xboxToken) { @@ -55,24 +60,44 @@ public NetherNetXboxRpcSignaling(String xboxToken) { @Override protected void onConnected(ChannelHandlerContext ctx) { - ctx.executor().scheduleAtFixedRate(() -> { - if (channel != null && channel.isActive()) { - sendJsonRpcRequest(NetherNetConstants.XBOX_RPC_METHOD_PING, new JsonObject()); - } - }, 30, 50, TimeUnit.SECONDS); + scheduleRecurring(ctx, "rpc-ping", () -> + sendJsonRpcRequest(NetherNetConstants.XBOX_RPC_METHOD_PING, new JsonObject()), 30, 50); + + scheduleRecurring(ctx, "turn-refresh", this::refreshTurnCredentials, + TURN_REFRESH_INTERVAL_SECONDS, TURN_REFRESH_INTERVAL_SECONDS); + refreshTurnCredentials(); + } + + /** + * Fetches TURN credentials and applies them via updateIceServers, which + * also completes the connect future during the initial exchange. On later + * refreshes a failure only logs; the previous credentials stay in place. + */ + private void refreshTurnCredentials() { sendJsonRpcRequest(NetherNetConstants.XBOX_RPC_METHOD_TURN_AUTH, new JsonObject()) - .thenAccept(response -> { - List servers = parseTurnServers(response); - if (connectFuture != null && !connectFuture.isDone()) connectFuture.complete(servers); - }) + .thenAccept(response -> updateIceServers(parseTurnServers(response))) .exceptionally(t -> { log.error("Failed to fetch TURN credentials", t); - if (connectFuture != null && !connectFuture.isDone()) connectFuture.completeExceptionally(t); + synchronized (this) { + if (connectFuture != null && !connectFuture.isDone()) connectFuture.completeExceptionally(t); + } return null; }); } + @Override + protected void onChannelInactive(ChannelHandlerContext ctx) { + // Fail everything that was waiting on a reply over the dead socket so + // callers see a prompt error instead of a future that never completes. + for (String id : pendingRequests.keySet()) { + CompletableFuture future = pendingRequests.remove(id); + if (future != null) { + future.completeExceptionally(new ClosedChannelException()); + } + } + } + @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) { String text = frame.text(); @@ -93,12 +118,12 @@ private void handleResponse(JsonObject json) { if (!json.has("id") || json.get("id").isJsonNull()) return; String id = json.get("id").getAsString(); CompletableFuture future = pendingRequests.remove(id); - + if (future != null) { if (json.has("error") && !json.get("error").isJsonNull()) { JsonObject error = json.getAsJsonObject("error"); String msg = error.has("message") ? error.get("message").getAsString() : error.toString(); - + boolean isNotFound = msg.contains("Player not registered"); if (!isNotFound && error.has("data") && error.get("data").isJsonObject()) { JsonObject data = error.getAsJsonObject("data"); @@ -161,6 +186,7 @@ private void processIncomingMessage(JsonObject msgObj) { @Override public void sendSignal(String targetNetworkId, String data) { + var channel = this.channel; if (channel == null || !channel.isActive()) throw new IllegalStateException("Signaling channel is not active"); JsonObject innerParams = new JsonObject(); @@ -192,9 +218,10 @@ private CompletableFuture sendJsonRpcRequest(String method, JsonObje rpc.addProperty("id", id); CompletableFuture future = new CompletableFuture<>(); - pendingRequests.put(id, future); + var channel = this.channel; if (channel != null && channel.isActive()) { + pendingRequests.put(id, future); channel.writeAndFlush(new TextWebSocketFrame(gson.toJson(rpc))); } else { future.completeExceptionally(new ClosedChannelException()); @@ -207,6 +234,7 @@ private void sendJsonRpcResult(JsonElement id, JsonElement result) { response.add("id", id); response.add("result", result); response.addProperty("jsonrpc", "2.0"); + var channel = this.channel; if (channel != null && channel.isActive()) channel.writeAndFlush(new TextWebSocketFrame(gson.toJson(response))); } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java index 0bd0072..3894c89 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java @@ -10,7 +10,6 @@ import java.net.URI; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; @Sharable public class NetherNetXboxSignaling extends AbstractNetherNetXboxSignaling { @@ -18,7 +17,7 @@ public class NetherNetXboxSignaling extends AbstractNetherNetXboxSignaling { /** * Creates a NetherNetXboxSignaling instance. - * + * * @param networkId The Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -28,7 +27,7 @@ public NetherNetXboxSignaling(String networkId, String xboxToken) { /** * Creates a NetherNetXboxSignaling instance. - * + * * @param localNetworkId The local Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -38,7 +37,7 @@ public NetherNetXboxSignaling(long localNetworkId, String xboxToken) { /** * Creates a NetherNetXboxSignaling instance with a random local Network ID. - * + * * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ public NetherNetXboxSignaling(String xboxToken) { @@ -47,28 +46,16 @@ public NetherNetXboxSignaling(String xboxToken) { @Override protected void onConnected(ChannelHandlerContext ctx) { - // IMPORTANT: scheduleAtFixedRate silently cancels the task on the first - // exception thrown by the runnable. A single transient write failure would - // kill the ping loop forever, leaving the connection to die to idle timeouts. - // We wrap in try/catch and add a write listener so failures are logged - // but the task keeps running on its regular cadence. - ctx.executor().scheduleAtFixedRate(() -> { - try { - if (!ctx.channel().isActive()) { - return; - } - JsonObject ping = new JsonObject(); - ping.addProperty("Type", 0); - ctx.writeAndFlush(new TextWebSocketFrame(gson.toJson(ping))) - .addListener(future -> { - if (!future.isSuccess()) { - log.warn("Ping write failed: {}", future.cause() != null ? future.cause().getMessage() : "unknown"); - } - }); - } catch (Throwable t) { - log.warn("Ping iteration threw; loop continues: {}", t.getMessage()); - } - }, 5, 5, TimeUnit.SECONDS); + scheduleRecurring(ctx, "app-ping", () -> { + JsonObject ping = new JsonObject(); + ping.addProperty("Type", 0); + ctx.writeAndFlush(new TextWebSocketFrame(gson.toJson(ping))) + .addListener(future -> { + if (!future.isSuccess()) { + log.warn("Ping write failed: {}", future.cause() != null ? future.cause().getMessage() : "unknown"); + } + }); + }, 5, 5); } @Override @@ -95,11 +82,15 @@ protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) } case NetherNetConstants.XBOX_SIGNAL_CREDENTIALS -> { log.trace("Received Credentials"); - if (json.has("Message") && connectFuture != null && !connectFuture.isDone()) { + if (json.has("Message")) { String rawMsg = json.get("Message").getAsString(); JsonObject credentials = JsonParser.parseString(rawMsg).getAsJsonObject(); - - connectFuture.complete(parseTurnServers(credentials)); + + // Applied unconditionally, not just while connecting: the + // service pushes refreshed TURN credentials over the + // lifetime of the socket, and peer connections created + // later must not be handed the expired originals. + updateIceServers(parseTurnServers(credentials)); } } case NetherNetConstants.XBOX_SIGNAL_ACCEPTED, NetherNetConstants.XBOX_SIGNAL_ACK -> log.trace("Signal Ack: {}", text); @@ -112,6 +103,7 @@ protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) @Override public void sendSignal(String targetNetworkId, String data) { + var channel = this.channel; if (channel != null && channel.isActive()) { JsonObject msg = new JsonObject(); msg.addProperty("Type", 1); @@ -122,4 +114,4 @@ public void sendSignal(String targetNetworkId, String data) { throw new IllegalStateException("Attempted to send signal to " + targetNetworkId + " but WebSocket is closed!"); } } -} \ No newline at end of file +} From 2131d7addc5285f0aa3217df612703514e62ac45 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:21:23 +0200 Subject: [PATCH 3/7] Optimize the data channel path isActive() reads an observer maintained flag instead of a blocking JNI call per write. Outbound fragmentation reuses one scratch direct buffer per channel and sends with sendAsync, which no longer stalls the event loop on the native network thread. Inbound packets are copied once instead of twice: complete single segment messages take a fast path and assembled multi segment buffers are handed off instead of re copied. --- .../channel/nethernet/NetherNetChannel.java | 120 ++++++++++++++---- 1 file changed, 97 insertions(+), 23 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java index c1ef4d2..4bdd882 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java @@ -36,6 +36,16 @@ public abstract class NetherNetChannel extends AbstractChannel { protected RTCDataChannel reliableChannel; protected RTCDataChannel unreliableChannel; + // Mirrors the reliable channel's OPEN state, updated from the data channel + // observer. isActive() is called on every write and formerly crossed JNI + // into a blocking native call for it; the cached flag makes it free. + private volatile boolean reliableOpen; + + // Reusable outbound fragment buffer. Confined to the event loop (doWrite), + // and safe to reuse across sends because the native layer copies the data + // out before sendAsync returns. + private ByteBuffer outboundScratch; + protected final Queue pendingWrites = new ConcurrentLinkedQueue<>(); private final AtomicBoolean channelActiveFired = new AtomicBoolean(); @@ -66,6 +76,8 @@ protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketA public void setDataChannels(RTCDataChannel reliable, RTCDataChannel unreliable) { this.reliableChannel = reliable; + // Seed the cached state once; the observer keeps it current from here. + this.reliableOpen = reliable.getState() == RTCDataChannelState.OPEN; // The unreliable channel is intentionally stored but never observed or // written to: this is a reliable-only implementation, which gives the // ordered, lossless stream a Bedrock translation proxy needs. If it is @@ -94,6 +106,9 @@ public void onStateChange() { @Override public void onMessage(RTCDataChannelBuffer buffer) { + // The ByteBuffer wraps native memory that is only valid for + // the duration of this callback, so every path below must copy + // exactly once into a netty buffer before handing off. ByteBuffer data = buffer.data; if (!data.hasRemaining()) return; @@ -105,6 +120,15 @@ public void onMessage(RTCDataChannelBuffer buffer) { return; } + // Fast path: complete single-segment message (the common + // case) skips the assembly buffer entirely. + if (currentSegmentCount == -1 && segments == 0) { + if (data.hasRemaining()) { + deliver(copyToBuf(data)); + } + return; + } + if (currentSegmentCount == -1) { currentSegmentCount = segments; } else { @@ -117,27 +141,17 @@ public void onMessage(RTCDataChannelBuffer buffer) { } if (data.hasRemaining()) { - byte[] payload = new byte[data.remaining()]; - data.get(payload); - assemblyBuf.writeBytes(payload); + assemblyBuf.writeBytes(data); } if (segments == 0) { - try { - if (assemblyBuf.isReadable()) { - ByteBuf packet = assemblyBuf.copy(); - - eventLoop().execute(() -> { - fireChannelActiveIfReady(); - pipeline().fireChannelRead(packet); - pipeline().fireChannelReadComplete(); - }); - } - } catch (Exception e) { - log.error("Error processing packet", e); - } finally { - assemblyBuf.clear(); - currentSegmentCount = -1; + currentSegmentCount = -1; + if (assemblyBuf.isReadable()) { + // Hand the assembled buffer off to the pipeline and + // start a fresh one instead of copying it again. + ByteBuf packet = assemblyBuf; + assemblyBuf = config.getAllocator().buffer(); + deliver(packet); } } } @@ -151,6 +165,33 @@ public void onMessage(RTCDataChannelBuffer buffer) { } } + /** + * Copies the remaining bytes of the given callback-scoped buffer into a + * freshly allocated netty buffer. + */ + private ByteBuf copyToBuf(ByteBuffer data) { + ByteBuf packet = config.getAllocator().buffer(data.remaining()); + packet.writeBytes(data); + return packet; + } + + /** + * Fires a fully reassembled packet down the pipeline on the event loop. + * Releases the packet if the event loop rejects the task (shutdown race) + * so the buffer cannot leak. + */ + private void deliver(ByteBuf packet) { + try { + eventLoop().execute(() -> { + fireChannelActiveIfReady(); + pipeline().fireChannelRead(packet); + pipeline().fireChannelReadComplete(); + }); + } catch (Exception e) { + packet.release(); + } + } + /** * Sets the maximum outbound SCTP message size, in bytes, for this channel. * Should be the {@code a=max-message-size} the remote peer advertised in its @@ -165,9 +206,18 @@ public void setMaxOutboundMessageSize(int size) { } private void onDataChannelStateChange() { + RTCDataChannel channel = this.reliableChannel; + if (channel == null) { + return; + } + // One native call per state transition; every other isActive() check + // reads the cached flag. + RTCDataChannelState state = channel.getState(); + reliableOpen = state == RTCDataChannelState.OPEN; + if (isActive()) { fireChannelActiveIfReady(); - } else if (reliableChannel.getState() == RTCDataChannelState.CLOSED) { + } else if (state == RTCDataChannelState.CLOSED) { close(); } } @@ -242,19 +292,25 @@ private void writeInternal(Object msg) { ByteBuf framed = payload.retainedDuplicate(); try { + // One reusable direct buffer for every fragment of every message: + // sendAsync copies the bytes out before returning, so the scratch + // can be refilled immediately, and unlike the blocking send it + // never stalls this event loop on the native network thread. + ByteBuffer chunk = outboundScratch(1 + maxPayload); + int offset = 0; for (int i = 0; i < segments; i++) { int remaining = segments - 1 - i; int chunkSize = Math.min(maxPayload, framed.readableBytes() - offset); - ByteBuffer chunk = ByteBuffer.allocateDirect(1 + chunkSize); + chunk.clear(); + chunk.limit(1 + chunkSize); chunk.put((byte) remaining); framed.getBytes(offset, chunk); - chunk.position(chunk.limit()); chunk.flip(); - reliableChannel.send(new RTCDataChannelBuffer(chunk, true)); + reliableChannel.sendAsync(new RTCDataChannelBuffer(chunk, true)); offset += chunkSize; } } catch (Exception e) { @@ -264,6 +320,20 @@ private void writeInternal(Object msg) { } } + /** + * Returns the reusable outbound fragment buffer, growing it when the + * negotiated max message size demands more. Only used from doWrite on the + * event loop. + */ + private ByteBuffer outboundScratch(int capacity) { + ByteBuffer scratch = this.outboundScratch; + if (scratch == null || scratch.capacity() < capacity) { + scratch = ByteBuffer.allocateDirect(capacity); + this.outboundScratch = scratch; + } + return scratch; + } + @Override protected void doRegister() throws Exception { if (isActive()) { @@ -288,6 +358,8 @@ protected void doDisconnect() throws Exception { @Override protected void doClose() throws Exception { this.open = false; + this.reliableOpen = false; + this.outboundScratch = null; if (reliableChannel != null) { reliableChannel.unregisterObserver(); @@ -350,7 +422,9 @@ public boolean isOpen() { @Override public boolean isActive() { - return isOpen() && this.reliableChannel != null && this.reliableChannel.getState() == RTCDataChannelState.OPEN; + // Hot path: called around every write. Reads the observer-maintained + // flag instead of a blocking JNI call into the native network thread. + return isOpen() && this.reliableChannel != null && this.reliableOpen; } @Override From 7af2f676189675dba93b8c3e0b7f9caad94bdb4f Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:21:23 +0200 Subject: [PATCH 4/7] Offload connection setup, pool factories, real remote addresses Connection setup hops from the signaling I/O thread to the server channel event loop: the signaling thread only registers the signal handler (so candidates arriving behind an offer are never dropped) and its keepalives can no longer be starved by blocking native WebRTC calls. Candidates arriving before the peer connection exists are queued in order. The server channel can be backed by a pool of PeerConnectionFactory instances with round robin assignment, spreading DTLS and SCTP load across several native network threads instead of serializing every player on one. Channels now receive their real remote address from the ICE selected candidate pair callback, which fires before DTLS and the data channels open, replacing the random placeholder before activation. Relayed connections record the TURN relay address. The old commented out getStats approach is removed. --- .../nethernet/NetherNetChannelFactory.java | 18 + .../nethernet/NetherNetServerChannel.java | 381 +++++++++++------- 2 files changed, 261 insertions(+), 138 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java index 82c3cc0..8860b54 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java @@ -6,6 +6,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelFactory; +import java.util.List; import java.util.function.Supplier; public class NetherNetChannelFactory implements ChannelFactory { @@ -32,6 +33,23 @@ public static ChannelFactory server(PeerConnectionFactor return new NetherNetChannelFactory<>(() -> new NetherNetServerChannel(factory, signaling)); } + /** + * Creates a NetherNet Server Channel Factory backed by a pool of + * PeerConnectionFactory instances. Each native factory carries exactly one + * network, worker, and signaling thread shared by every peer connection it + * creates, so a pool multiplies the threads available to the data plane: + * incoming connections are assigned round robin across the pool. + * + * @param factories The PeerConnectionFactory pool. Must not be empty. The + * resulting server channel takes ownership and disposes + * every factory on close. + * @param signaling The NetherNetServerSignaling instance for signaling. + * @return A ChannelFactory for NetherNetServerChannel. + */ + public static ChannelFactory server(List factories, NetherNetServerSignaling signaling) { + return new NetherNetChannelFactory<>(() -> new NetherNetServerChannel(factories, signaling)); + } + /** * Creates a NetherNet Client Channel Factory. * diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java index fc8480e..f66e09b 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java @@ -28,6 +28,7 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -37,15 +38,19 @@ public class NetherNetServerChannel extends AbstractServerChannel { private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16); private final DefaultNetherServerChannelConfig config; - private final PeerConnectionFactory factory; + private final List factories; private final NetherNetServerSignaling signaling; - + + // Round robin cursor into factories. Only touched from this channel's + // event loop (establishConnection), so no synchronization is needed. + private int nextFactory; + private InetSocketAddress localAddress; private volatile boolean open = true; /** * Creates a NetherNetServerChannel with a new PeerConnectionFactory. - * + * * @param signaling The NetherNetServerSignaling instance for signaling. */ public NetherNetServerChannel(NetherNetServerSignaling signaling) { @@ -54,12 +59,30 @@ public NetherNetServerChannel(NetherNetServerSignaling signaling) { /** * Creates a NetherNetServerChannel. - * + * * @param factory The PeerConnectionFactory to use for creating peer connections. Should be reused where possible. * @param signaling The NetherNetServerSignaling instance for signaling. */ public NetherNetServerChannel(PeerConnectionFactory factory, NetherNetServerSignaling signaling) { - this.factory = factory; + this(List.of(factory), signaling); + } + + /** + * Creates a NetherNetServerChannel backed by a pool of PeerConnectionFactory + * instances. Each native factory runs one network, worker, and signaling + * thread shared by all its peer connections; connections are assigned to + * factories round robin, spreading the DTLS and SCTP load of many players + * across the pool instead of serializing it on a single network thread. + * + * @param factories The PeerConnectionFactory pool, at least one. This + * channel takes ownership and disposes each on close. + * @param signaling The NetherNetServerSignaling instance for signaling. + */ + public NetherNetServerChannel(List factories, NetherNetServerSignaling signaling) { + if (factories.isEmpty()) { + throw new IllegalArgumentException("factories must not be empty"); + } + this.factories = List.copyOf(factories); this.signaling = signaling; this.config = new DefaultNetherServerChannelConfig(this); } @@ -68,7 +91,7 @@ public NetherNetServerChannel(PeerConnectionFactory factory, NetherNetServerSign protected void doBind(SocketAddress localAddress) throws Exception { if (!(localAddress instanceof InetSocketAddress)) throw new IllegalArgumentException("Unsupported address type"); this.localAddress = (InetSocketAddress) localAddress; - + this.signaling.setNewConnectionHandler((connectionId, remoteNetworkId, offerSdp) -> { acceptConnection(connectionId, offerSdp, remoteNetworkId); }); @@ -76,48 +99,151 @@ protected void doBind(SocketAddress localAddress) throws Exception { this.signaling.bind(localAddress); } + /** + * Accepts an incoming connection offer. Runs on the signaling I/O thread: + * only the signal handler registration happens here (a cheap map put, so + * candidates arriving right behind the offer are never dropped), then all + * WebRTC work hops onto this server channel's event loop. Native peer + * connection calls block on the WebRTC signaling thread and must never + * stall the signaling socket's thread, whose keepalives are what hold the + * connection to the signaling service open. + */ public void acceptConnection(long connectionId, String offerSdp, String remoteNetworkId) { - RTCConfiguration rtcConfig = new RTCConfiguration(); - rtcConfig.portAllocatorConfig = this.config.getOption(NetherChannelOption.NETHER_PORT_ALLOCATOR_CONFIG); - rtcConfig.bundlePolicy = RTCBundlePolicy.MAX_BUNDLE; - - // Inject ICE servers if the signaling implementation supports it - List iceServers = this.signaling.getIceServers(); - if (iceServers != null && !iceServers.isEmpty()) { - log.trace("Injecting {} ICE Servers into PeerConnection for {}", iceServers.size(), Long.toUnsignedString(connectionId)); - for (IceServerInfo info : iceServers) { - RTCIceServer iceServer = new RTCIceServer(); - iceServer.urls = info.urls(); - iceServer.username = info.username(); - iceServer.password = info.password(); - rtcConfig.iceServers.add(iceServer); + PendingConnection pending = new PendingConnection(connectionId); + signaling.setSignalHandler(connectionId, signal -> eventLoop().execute(() -> pending.handleSignal(signal))); + eventLoop().execute(() -> establishConnection(pending, connectionId, offerSdp, remoteNetworkId)); + } + + /** + * Builds the peer connection for an accepted offer. Runs on this server + * channel's event loop; incoming signals for the connection hop onto the + * same loop, so everything here is single-threaded and ordered. + */ + private void establishConnection(PendingConnection pending, long connectionId, String offerSdp, String remoteNetworkId) { + try { + RTCConfiguration rtcConfig = new RTCConfiguration(); + rtcConfig.portAllocatorConfig = this.config.getOption(NetherChannelOption.NETHER_PORT_ALLOCATOR_CONFIG); + rtcConfig.bundlePolicy = RTCBundlePolicy.MAX_BUNDLE; + + // Inject ICE servers if the signaling implementation supports it + List iceServers = this.signaling.getIceServers(); + if (iceServers != null && !iceServers.isEmpty()) { + log.trace("Injecting {} ICE Servers into PeerConnection for {}", iceServers.size(), Long.toUnsignedString(connectionId)); + for (IceServerInfo info : iceServers) { + RTCIceServer iceServer = new RTCIceServer(); + iceServer.urls = info.urls(); + iceServer.username = info.username(); + iceServer.password = info.password(); + rtcConfig.iceServers.add(iceServer); + } } + + ServerPeerConnectionObserver observer = new ServerPeerConnectionObserver(connectionId, remoteNetworkId); + PeerConnectionFactory factory = factories.get(nextFactory); + nextFactory = (nextFactory + 1) % factories.size(); + RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, observer); + + NetherNetChildChannel child = new NetherNetChildChannel(this, pc, generatePlaceholderAddress(), localAddress); + // Fragment outbound data no larger than the client advertised it can + // receive (a=max-message-size in its offer), falling back to the + // conservative default when the client does not advertise one. + child.setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(offerSdp, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); + observer.setChildChannel(child); + + child.closeFuture().addListener(future -> signaling.removeSignalHandler(connectionId)); + + int handshakeTimeoutSeconds = this.config.getOption(NetherChannelOption.NETHER_SERVER_RTC_HANDSHAKE_TIMEOUT_SECONDS); + ScheduledFuture timeoutTask = eventLoop().schedule(() -> { + if (!child.isActive()) { + log.warn("Connection {} timed out during handshake ({}s)", Long.toUnsignedString(connectionId), handshakeTimeoutSeconds); + child.close(); + pc.close(); + } + }, handshakeTimeoutSeconds, TimeUnit.SECONDS); + observer.setHandshakeTimeout(timeoutTask); + + // Handle Offer + pc.setRemoteDescription(new RTCSessionDescription(RTCSdpType.OFFER, offerSdp), new SetSessionDescriptionObserver() { + @Override + public void onSuccess() { + log.trace("Remote description set for {}", Long.toUnsignedString(connectionId)); + pc.createAnswer(new RTCAnswerOptions(), new CreateSessionDescriptionObserver() { + @Override + public void onSuccess(RTCSessionDescription description) { + pc.setLocalDescription(description, new SetSessionDescriptionObserver() { + @Override + public void onSuccess() { + log.trace("Sending Answer SDP for {}", Long.toUnsignedString(connectionId)); + try { + signaling.sendSignal( + remoteNetworkId, + NetherNetConstants.buildSignalConnectResponse(connectionId, description.sdp) + ); + } catch (Exception e) { + // Signaling dropped mid-handshake; the client cannot + // receive the answer, so let the handshake timeout + // reap this connection. + log.warn("Failed to send answer for {} (signaling unavailable): {}", + Long.toUnsignedString(connectionId), e.getMessage()); + return; + } + pipeline().fireChannelRead(child); + } + @Override public void onFailure(String error) { log.error("SetLocalDesc failed: {}", error); } + }); + } + @Override public void onFailure(String error) { log.error("CreateAnswer failed: {}", error); } + }); + } + @Override public void onFailure(String error) { log.error("SetRemoteDesc failed: {}", error); } + }); + + // Publish the peer connection to the signal handler and drain any + // candidates that arrived while it was being created. + pending.attach(pc, child); + } catch (Exception e) { + log.error("Failed to establish connection {}: {}", Long.toUnsignedString(connectionId), e.getMessage(), e); + signaling.removeSignalHandler(connectionId); } + } - ServerPeerConnectionObserver observer = new ServerPeerConnectionObserver(connectionId, remoteNetworkId); - RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, observer); + /** + * Per-connection signal state. All methods run on this server channel's + * event loop, so no synchronization is needed. Signals that arrive between + * the offer and the peer connection becoming available are queued and + * drained by attach, preserving arrival order. + */ + private final class PendingConnection { + private final long connectionId; + private RTCPeerConnection pc; + private NetherNetChildChannel child; + private List queued = new ArrayList<>(); - NetherNetChildChannel child = new NetherNetChildChannel(this, pc, generatePlaceholderAddress(), localAddress); - // Fragment outbound data no larger than the client advertised it can - // receive (a=max-message-size in its offer), falling back to the - // conservative default when the client does not advertise one. - child.setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(offerSdp, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); - observer.setChildChannel(child); + PendingConnection(long connectionId) { + this.connectionId = connectionId; + } - child.closeFuture().addListener(future -> signaling.removeSignalHandler(connectionId)); + void attach(RTCPeerConnection pc, NetherNetChildChannel child) { + this.pc = pc; + this.child = child; + List pendingSignals = this.queued; + this.queued = null; + for (String signal : pendingSignals) { + apply(signal); + } + } - int handshakeTimeoutSeconds = this.config.getOption(NetherChannelOption.NETHER_SERVER_RTC_HANDSHAKE_TIMEOUT_SECONDS); - ScheduledFuture timeoutTask = eventLoop().schedule(() -> { - if (!child.isActive()) { - log.warn("Connection {} timed out during handshake ({}s)", Long.toUnsignedString(connectionId), handshakeTimeoutSeconds); - child.close(); - pc.close(); + void handleSignal(String signal) { + if (pc == null) { + if (queued != null) { + queued.add(signal); + } + return; } - }, handshakeTimeoutSeconds, TimeUnit.SECONDS); - observer.setHandshakeTimeout(timeoutTask); - - // Register Signal Handler - signaling.setSignalHandler(connectionId, (signal) -> { + apply(signal); + } + + private void apply(String signal) { String[] parts = signal.split(" ", 3); if (parts.length < 3) return; String type = parts[0]; @@ -137,34 +263,7 @@ public void acceptConnection(long connectionId, String offerSdp, String remoteNe child.close(); } } - }); - - // Handle Offer - pc.setRemoteDescription(new RTCSessionDescription(RTCSdpType.OFFER, offerSdp), new SetSessionDescriptionObserver() { - @Override - public void onSuccess() { - log.trace("Remote description set for {}", Long.toUnsignedString(connectionId)); - pc.createAnswer(new RTCAnswerOptions(), new CreateSessionDescriptionObserver() { - @Override - public void onSuccess(RTCSessionDescription description) { - pc.setLocalDescription(description, new SetSessionDescriptionObserver() { - @Override - public void onSuccess() { - log.trace("Sending Answer SDP for {}", Long.toUnsignedString(connectionId)); - signaling.sendSignal( - remoteNetworkId, - NetherNetConstants.buildSignalConnectResponse(connectionId, description.sdp) - ); - pipeline().fireChannelRead(child); - } - @Override public void onFailure(String error) { log.error("SetLocalDesc failed: {}", error); } - }); - } - @Override public void onFailure(String error) { log.error("CreateAnswer failed: {}", error); } - }); - } - @Override public void onFailure(String error) { log.error("SetRemoteDesc failed: {}", error); } - }); + } } /** @@ -174,13 +273,17 @@ private class ServerPeerConnectionObserver implements PeerConnectionObserver { private final long connectionId; private final String remoteNetworkId; private NetherNetChildChannel child; - + private RTCDataChannel reliable; private RTCDataChannel unreliable; private boolean dataChannelsSet; private ScheduledFuture handshakeTimeout; + // Real remote address from ICE nomination. Buffered here when the + // callback fires before the child channel is attached. + private volatile InetSocketAddress pendingRemoteAddress; + public ServerPeerConnectionObserver(long connectionId, String remoteNetworkId) { this.connectionId = connectionId; this.remoteNetworkId = remoteNetworkId; @@ -192,19 +295,57 @@ public void setHandshakeTimeout(ScheduledFuture handshakeTimeout) { public void setChildChannel(NetherNetChildChannel child) { this.child = child; + InetSocketAddress pending = this.pendingRemoteAddress; + if (pending != null) { + child.remoteAddress = pending; + } checkDataChannels(); } + @Override + public void onSelectedCandidatePairChanged(String remoteAddress, int remotePort, String candidateType) { + // ICE nominated a pair. This fires before DTLS and the data + // channels open, so the channel carries its real remote address + // before it activates and anything downstream reads it. Until + // then the channel holds its unique random placeholder, which + // also covers the case of this callback never firing. For a + // relayed connection the address is the TURN relay, which is the + // peer actually connected to us and the intended value. A later + // re-nomination simply overwrites. + try { + InetSocketAddress resolved = new InetSocketAddress(remoteAddress, remotePort); + NetherNetChildChannel target = this.child; + if (target != null) { + target.remoteAddress = resolved; + } else { + this.pendingRemoteAddress = resolved; + } + log.debug("Resolved remote address for {}: {}:{} (type: {})", + Long.toUnsignedString(this.connectionId), remoteAddress, remotePort, candidateType); + } catch (Exception e) { + log.debug("Failed to set remote address for {}: {}", + Long.toUnsignedString(this.connectionId), e.getMessage()); + } + } + @Override public void onIceCandidate(RTCIceCandidate candidate) { if (log.isTraceEnabled()) { - log.trace("Generated ICE Candidate for {}: {} (Type: {})", + log.trace("Generated ICE Candidate for {}: {} (Type: {})", Long.toUnsignedString(this.connectionId), candidate.sdp, extractCandidateType(candidate.sdp)); } - signaling.sendSignal( - remoteNetworkId, - NetherNetConstants.buildSignalCandidateAdd(connectionId, candidate.sdp) - ); + try { + signaling.sendSignal( + remoteNetworkId, + NetherNetConstants.buildSignalCandidateAdd(connectionId, candidate.sdp) + ); + } catch (Exception e) { + // Signaling dropped mid-handshake. Established connections don't + // signal candidates, so only this in-flight handshake is affected; + // the handshake timeout cleans it up if it cannot complete. + log.debug("Failed to signal ICE candidate for {} (signaling unavailable): {}", + Long.toUnsignedString(this.connectionId), e.getMessage()); + } } private String extractCandidateType(String sdp) { @@ -232,16 +373,16 @@ public void onConnectionChange(RTCPeerConnectionState state) { public void onDataChannel(RTCDataChannel dataChannel) { String label = dataChannel.getLabel(); log.debug("Received Data Channel: {}", label); - + if (NetherNetConstants.RELIABLE_CHANNEL_LABEL.equals(label)) { this.reliable = dataChannel; } else if (NetherNetConstants.UNRELIABLE_CHANNEL_LABEL.equals(label)) { this.unreliable = dataChannel; } - + checkDataChannels(); } - + private void checkDataChannels() { if (child != null && reliable != null && unreliable != null && !dataChannelsSet) { dataChannelsSet = true; @@ -255,54 +396,6 @@ private void checkDataChannels() { } } - // TODO: Resolve real remote address from ICE candidate pair via webrtc-java fork. - // The code below works but delays fireChannelActive, which causes pipeline timing - // issues (duplicate handler registration). Requires a synchronous getter in the - // native JNI layer to avoid the delay. Until then, connections use a unique - // 10.x.x.x placeholder address set at channel creation time. - // - // private void resolveRemoteAddressThenActivate(NetherNetChildChannel child) { - // java.util.concurrent.atomic.AtomicBoolean activated = new java.util.concurrent.atomic.AtomicBoolean(false); - // ScheduledFuture fallback = eventLoop().schedule(() -> { - // activateOnce(child, activated); - // }, 2, TimeUnit.SECONDS); - // try { - // child.peerConnection.getStats(report -> { - // try { - // String remoteId = null; - // for (var entry : report.getStats().entrySet()) { - // var stats = entry.getValue(); - // if (stats.getType() == dev.kastle.webrtc.RTCStatsType.CANDIDATE_PAIR) { - // Object selected = stats.getAttributes().get("nominated"); - // if (Boolean.TRUE.equals(selected)) { - // remoteId = (String) stats.getAttributes().get("remoteCandidateId"); - // break; - // } - // } - // } - // if (remoteId != null) { - // var remoteStat = report.getStats().get(remoteId); - // if (remoteStat != null) { - // String ip = (String) remoteStat.getAttributes().get("ip"); - // Object portObj = remoteStat.getAttributes().get("port"); - // if (ip != null && portObj != null) { - // int port = (portObj instanceof Number) ? ((Number) portObj).intValue() : Integer.parseInt(portObj.toString()); - // child.remoteAddress = new InetSocketAddress(ip, port); - // } - // } - // } - // } catch (Exception e) { - // log.debug("Failed to resolve remote address: {}", e.getMessage()); - // } finally { - // fallback.cancel(false); - // activateOnce(child, activated); - // } - // }); - // } catch (Exception e) { - // fallback.cancel(false); - // activateOnce(child, activated); - // } - // } // // private void activateOnce(NetherNetChildChannel child, java.util.concurrent.atomic.AtomicBoolean activated) { // if (!activated.compareAndSet(false, true)) return; @@ -328,11 +421,23 @@ private static InetSocketAddress generatePlaceholderAddress() { @Override protected void doClose() throws Exception { this.open = false; - + try { signaling.close(); } finally { - factory.dispose(); + Exception failure = null; + for (PeerConnectionFactory factory : factories) { + try { + factory.dispose(); + } catch (Exception e) { + // Keep disposing the rest; rethrow the first failure after. + if (failure == null) failure = e; + log.warn("Failed to dispose PeerConnectionFactory: {}", e.getMessage()); + } + } + if (failure != null) { + throw failure; + } } } @@ -348,24 +453,24 @@ protected SocketAddress localAddress0() { @Override protected boolean isCompatible(EventLoop loop) { - return true; + return true; } @Override public ChannelConfig config() { return config; } - - @Override - public boolean isOpen() { + + @Override + public boolean isOpen() { return this.open; } - - @Override - public boolean isActive() { + + @Override + public boolean isActive() { return isOpen() && localAddress0() != null; } - - @Override - public ChannelMetadata metadata() { - return METADATA; + + @Override + public ChannelMetadata metadata() { + return METADATA; } } From 79ed4909379e900264d74f9db89c218966a1f415 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Fri, 3 Jul 2026 09:49:36 +0200 Subject: [PATCH 5/7] Scope pending RPC request failures to their own socket Each in flight JSON RPC request now records the WebSocket channel it was written to. When a socket dies, onChannelInactive fails only the requests that were sent on that socket: during a reconnect the old channel's inactive event can no longer fail requests already written to the replacement socket, which previously left those callers with a spurious ClosedChannelException while the reply was still on its way. --- .../signaling/NetherNetXboxRpcSignaling.java | 37 +++++++++++++++---- 1 file changed, 29 insertions(+), 8 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java index 21ce617..49b346e 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java @@ -7,6 +7,7 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import dev.kastle.netty.channel.nethernet.NetherNetConstants; +import io.netty.channel.Channel; import io.netty.channel.ChannelHandler.Sharable; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; @@ -27,7 +28,23 @@ public class NetherNetXboxRpcSignaling extends AbstractNetherNetXboxSignaling { * connections created on a long-lived socket never receive expired ones. */ private static final long TURN_REFRESH_INTERVAL_SECONDS = 30 * 60; - private final Map> pendingRequests = new ConcurrentHashMap<>(); + /** + * An in-flight JSON-RPC request, tagged with the WebSocket channel it was + * written to so that a dying channel only fails its own requests. During + * a reconnect the old channel's inactive event must not fail requests + * already sent on the replacement socket. + */ + private static final class PendingRequest { + final CompletableFuture future; + final Channel channel; + + PendingRequest(CompletableFuture future, Channel channel) { + this.future = future; + this.channel = channel; + } + } + + private final Map pendingRequests = new ConcurrentHashMap<>(); /** * Creates a NetherNetXboxRpcSignaling instance. @@ -90,12 +107,15 @@ private void refreshTurnCredentials() { protected void onChannelInactive(ChannelHandlerContext ctx) { // Fail everything that was waiting on a reply over the dead socket so // callers see a prompt error instead of a future that never completes. - for (String id : pendingRequests.keySet()) { - CompletableFuture future = pendingRequests.remove(id); - if (future != null) { - future.completeExceptionally(new ClosedChannelException()); + // Only requests written to THIS channel: during a reconnect the old + // channel's inactive event must not fail the new socket's requests. + pendingRequests.entrySet().removeIf(entry -> { + if (entry.getValue().channel == ctx.channel()) { + entry.getValue().future.completeExceptionally(new ClosedChannelException()); + return true; } - } + return false; + }); } @Override @@ -117,7 +137,8 @@ protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) private void handleResponse(JsonObject json) { if (!json.has("id") || json.get("id").isJsonNull()) return; String id = json.get("id").getAsString(); - CompletableFuture future = pendingRequests.remove(id); + PendingRequest pending = pendingRequests.remove(id); + CompletableFuture future = pending != null ? pending.future : null; if (future != null) { if (json.has("error") && !json.get("error").isJsonNull()) { @@ -221,7 +242,7 @@ private CompletableFuture sendJsonRpcRequest(String method, JsonObje var channel = this.channel; if (channel != null && channel.isActive()) { - pendingRequests.put(id, future); + pendingRequests.put(id, new PendingRequest(future, channel)); channel.writeAndFlush(new TextWebSocketFrame(gson.toJson(rpc))); } else { future.completeExceptionally(new ClosedChannelException()); From 4ee49e643a8eb04a4981aec9b04736309be94325 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Fri, 3 Jul 2026 10:17:59 +0200 Subject: [PATCH 6/7] Scope client handshake retries to their attempt Every retry bumps a generation counter that async engine callbacks capture: offer creation, description observers, ICE candidate emission, connection state changes, and the data channel open handler all bail once a retry has moved past them. Without this a delayed callback from an abandoned attempt could send stale SDP or candidates under the new connection id, trigger a spurious extra retry, or mark the handshake complete with data channels belonging to a closed peer connection. The signal id check also runs again inside the event loop task, since a retry can cycle the connection id between the check on the signaling thread and the task running, which previously let an old attempt's answer reach the new attempt's peer connection. --- .../nethernet/NetherNetClientChannel.java | 79 +++++++++++++++---- 1 file changed, 62 insertions(+), 17 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java index fd5e463..70081fb 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java @@ -18,6 +18,7 @@ import dev.kastle.webrtc.RTCIceCandidate; import dev.kastle.webrtc.RTCIceServer; import dev.kastle.webrtc.RTCOfferOptions; +import dev.kastle.webrtc.RTCPeerConnection; import dev.kastle.webrtc.RTCPeerConnectionState; import dev.kastle.webrtc.RTCSdpType; import dev.kastle.webrtc.RTCSessionDescription; @@ -53,6 +54,13 @@ public class NetherNetClientChannel extends NetherNetChannel { private int retryCount = 0; + // Monotonic attempt marker, bumped on every handshake retry. Async engine + // callbacks belonging to a previous attempt (offer creation, description + // observers, data channel state changes) capture their generation and + // bail once a retry has moved past them, so a delayed stale callback can + // no longer mutate the replacement attempt's state. + private volatile int attemptGeneration; + /** * Creates a NetherNetClientChannel with a new PeerConnectionFactory. * @@ -189,6 +197,7 @@ private void resetAndRetryHandshake() { } retryCount++; + attemptGeneration++; if (peerConnection != null) { peerConnection.close(); @@ -215,17 +224,27 @@ private void initWebRTC(List iceServers) { } } - peerConnection = factory.createPeerConnection(rtcConfig, new PeerConnectionObserver() { + final int gen = attemptGeneration; + final long attemptConnectionId = this.connectionId; + + RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, new PeerConnectionObserver() { @Override public void onIceCandidate(RTCIceCandidate candidate) { + if (gen != attemptGeneration) { + return; + } try { signaling.sendSignal( - targetNetworkId, - NetherNetConstants.buildSignalCandidateAdd(connectionId, candidate.sdp) + targetNetworkId, + NetherNetConstants.buildSignalCandidateAdd(attemptConnectionId, candidate.sdp) ); } catch (Exception e) { log.error("Failed to send ICE candidate", e); - eventLoop().execute(() -> resetAndRetryHandshake()); + eventLoop().execute(() -> { + if (gen == attemptGeneration) { + resetAndRetryHandshake(); + } + }); } } @@ -234,7 +253,11 @@ public void onConnectionChange(RTCPeerConnectionState state) { if (state == RTCPeerConnectionState.FAILED) { // Fast fail trigger: retry immediately instead of waiting for timeout log.warn("PeerConnection entered FAILED state, resetting and retrying handshake."); - eventLoop().execute(() -> resetAndRetryHandshake()); + eventLoop().execute(() -> { + if (gen == attemptGeneration) { + resetAndRetryHandshake(); + } + }); } else { log.trace("PeerConnection state changed to {}", state); } @@ -242,27 +265,36 @@ public void onConnectionChange(RTCPeerConnectionState state) { @Override public void onDataChannel(RTCDataChannel dataChannel) { } }); + this.peerConnection = pc; - setupDataChannels(); + setupDataChannels(pc, gen); } private void createAndSendOffer() { - if (peerConnection == null) return; - peerConnection.createOffer(new RTCOfferOptions(), new CreateSessionDescriptionObserver() { + final RTCPeerConnection pc = this.peerConnection; + final int gen = attemptGeneration; + final long attemptConnectionId = this.connectionId; + if (pc == null) return; + pc.createOffer(new RTCOfferOptions(), new CreateSessionDescriptionObserver() { @Override public void onSuccess(RTCSessionDescription description) { - if (peerConnection == null) return; - peerConnection.setLocalDescription(description, new SetSessionDescriptionObserver() { + if (gen != attemptGeneration) return; + pc.setLocalDescription(description, new SetSessionDescriptionObserver() { @Override public void onSuccess() { + if (gen != attemptGeneration) return; try { signaling.sendSignal( - targetNetworkId, - NetherNetConstants.buildSignalConnectRequest(connectionId, description.sdp) + targetNetworkId, + NetherNetConstants.buildSignalConnectRequest(attemptConnectionId, description.sdp) ); } catch (Exception e) { log.error("Failed to send Connect Request", e); - eventLoop().execute(() -> resetAndRetryHandshake()); + eventLoop().execute(() -> { + if (gen == attemptGeneration) { + resetAndRetryHandshake(); + } + }); } } @Override public void onFailure(String error) { /* Retry handled by timeout */ } @@ -280,8 +312,9 @@ private void handleSignal(String signal) { String data = parts.length > 2 ? parts[2] : ""; // Verify this signal belongs to the current attempt + final long signalId; try { - long signalId = Long.parseUnsignedLong(idStr); + signalId = Long.parseUnsignedLong(idStr); if (signalId != this.connectionId) { log.debug("Ignored stale signal for ID {}", idStr); return; @@ -291,6 +324,15 @@ private void handleSignal(String signal) { } eventLoop().execute(() -> { + // Re-validate on the event loop: a retry may have cycled the + // connection id between the check above (signaling thread) and + // this task running. Inside the task the id, generation, and + // peer connection mutate together, so passing this check means + // everything read below belongs to the current attempt. + if (signalId != this.connectionId) { + log.debug("Ignored stale signal for ID {} (attempt retried)", idStr); + return; + } if (peerConnection == null) return; if (!isOpen() || handshakeComplete) return; @@ -321,7 +363,7 @@ private void handleSignal(String signal) { }); } - private void setupDataChannels() { + private void setupDataChannels(RTCPeerConnection pc, int gen) { RTCDataChannelInit reliableInit = new RTCDataChannelInit(); reliableInit.ordered = true; reliableInit.protocol = NetherNetConstants.RELIABLE_CHANNEL_LABEL; @@ -330,14 +372,17 @@ private void setupDataChannels() { unreliableInit.ordered = false; unreliableInit.maxRetransmits = 0; - RTCDataChannel reliable = peerConnection.createDataChannel(NetherNetConstants.RELIABLE_CHANNEL_LABEL, reliableInit); - RTCDataChannel unreliable = peerConnection.createDataChannel(NetherNetConstants.UNRELIABLE_CHANNEL_LABEL, unreliableInit); + RTCDataChannel reliable = pc.createDataChannel(NetherNetConstants.RELIABLE_CHANNEL_LABEL, reliableInit); + RTCDataChannel unreliable = pc.createDataChannel(NetherNetConstants.UNRELIABLE_CHANNEL_LABEL, unreliableInit); reliable.registerObserver(new RTCDataChannelObserver() { @Override public void onStateChange() { if (reliable.getState() == RTCDataChannelState.OPEN) { eventLoop().execute(() -> { + if (gen != attemptGeneration) { + return; + } if (!handshakeComplete) { log.debug("NetherNet Connection Established!"); handshakeComplete = true; From 3fbf5f156b24a0ae800055adc303699d5d3d9534 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Fri, 3 Jul 2026 10:18:44 +0200 Subject: [PATCH 7/7] Buffer remote candidates until the answer is applied The native addIceCandidate rejects candidates while the peer connection has no remote description, and setRemoteDescription completes asynchronously, so a candidate arriving right behind the CONNECT_RESPONSE could be dropped. Candidates now queue until the remote description observer reports success and are then applied in arrival order. A handshake retry clears the queue along with the rest of the attempt state. --- .../nethernet/NetherNetClientChannel.java | 48 ++++++++++++++++++- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java index 70081fb..6c5c05a 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java @@ -33,6 +33,7 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.channels.ClosedChannelException; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -61,6 +62,13 @@ public class NetherNetClientChannel extends NetherNetChannel { // no longer mutate the replacement attempt's state. private volatile int attemptGeneration; + // Event loop confined. The synchronous native addIceCandidate rejects + // candidates applied while the peer connection has no remote description + // yet, so candidates arriving before the CONNECT_RESPONSE answer has been + // applied are buffered and drained once it succeeds. + private boolean remoteDescriptionSet; + private List pendingRemoteCandidates = new ArrayList<>(); + /** * Creates a NetherNetClientChannel with a new PeerConnectionFactory. * @@ -206,6 +214,8 @@ private void resetAndRetryHandshake() { signaling.removeSignalHandler(this.connectionId); this.cycleConnectionId(); + remoteDescriptionSet = false; + pendingRemoteCandidates = new ArrayList<>(); startHandshake(); } @@ -341,13 +351,30 @@ private void handleSignal(String signal) { // Fragment outbound data no larger than the remote advertised // it can receive (a=max-message-size in its answer). setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(data, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); + final int gen = attemptGeneration; peerConnection.setRemoteDescription(new RTCSessionDescription(RTCSdpType.ANSWER, data), new SetSessionDescriptionObserver() { - @Override public void onSuccess() {} + @Override public void onSuccess() { + // Apply candidates that arrived before the answer + // finished applying, in arrival order. + eventLoop().execute(() -> { + if (gen != attemptGeneration) return; + remoteDescriptionSet = true; + List drained = pendingRemoteCandidates; + pendingRemoteCandidates = new ArrayList<>(); + for (String candidate : drained) { + applyRemoteCandidate(candidate); + } + }); + } @Override public void onFailure(String e) { /* Retry handled by timeout */ } }); } case NetherNetConstants.RTC_NEGOTIATION_CANDIDATE_ADD -> { - peerConnection.addIceCandidate(new RTCIceCandidate("0", 0, data)); + if (remoteDescriptionSet) { + applyRemoteCandidate(data); + } else { + pendingRemoteCandidates.add(data); + } } case NetherNetConstants.RTC_NEGOTIATION_CONNECT_ERROR -> { log.error("Received SIGNAL_CONNECT_ERROR for {}.", Long.toUnsignedString(this.connectionId)); @@ -408,6 +435,23 @@ public void onStateChange() { }); } + /** + * Applies a remote ICE candidate to the current peer connection. Runs on + * the event loop, either directly from a CANDIDATE_ADD signal or from the + * drain after the remote description applies. + */ + private void applyRemoteCandidate(String candidateSdp) { + RTCPeerConnection pc = this.peerConnection; + if (pc == null) { + return; + } + try { + pc.addIceCandidate(new RTCIceCandidate("0", 0, candidateSdp)); + } catch (Exception e) { + log.debug("Failed to apply ICE candidate (connection likely closed): {}", e.toString()); + } + } + private long cycleConnectionId() { this.connectionId = ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE); return this.connectionId;