From a2f8b53c6478898cadba5c0f106bc55ce8c4eb0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Thu, 23 Jul 2026 15:35:28 +0800 Subject: [PATCH 1/5] fix: avoid KeepAlive NPE when idle fires before WS handshake IdleStateHandler can emit IdleStateEvent after TCP connect but before WebSocket handshake completes. KeepAliveHandler previously dereferenced a null channel field, logging "connection operation failed" and leaving half-dead connections that recovered only after pool retry. Also close the channel in exceptionCaught so the session pool reconnects promptly. Fixes customer intermittent Stream disconnect (Aone 84622857). Co-authored-by: Cursor --- .../app-stream-network-ws/pom.xml | 5 +++ .../stream/network/ws/KeepAliveHandler.java | 31 +++++++++++--- .../stream/network/ws/NettyClientHandler.java | 4 ++ .../network/ws/KeepAliveHandlerTest.java | 42 +++++++++++++++++++ 4 files changed, 76 insertions(+), 6 deletions(-) create mode 100644 app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java diff --git a/app-stream-network/app-stream-network-ws/pom.xml b/app-stream-network/app-stream-network-ws/pom.xml index b045dae..48dd5e5 100644 --- a/app-stream-network/app-stream-network-ws/pom.xml +++ b/app-stream-network/app-stream-network-ws/pom.xml @@ -30,5 +30,10 @@ io.netty netty-handler-proxy + + junit + junit + test + diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java index b2fa344..5f8b81a 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java @@ -31,8 +31,8 @@ public class KeepAliveHandler extends SimpleChannelInboundHandler { private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(KeepAliveHandler.class); private final Duration timeout; - private final static HashedWheelTimer TIMER = new HashedWheelTimer(); - private Channel channel; + private final static HashedWheelTimer TIMER = new HashedWheelTimer(); + private volatile Channel channel; private final Map timeouts; private final AtomicBoolean active; @@ -52,7 +52,15 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc } if (evt instanceof IdleStateEvent) { - channel.eventLoop().execute(new PingTask()); + // IdleStateHandler starts counting after TCP connect, before WS handshake. + // Never touch the channel field until handshake completes, otherwise NPE: + // connection operation failed (KeepAliveHandler.java:56) + if (active.get()) { + final Channel ch = channel != null ? channel : ctx.channel(); + if (ch != null && ch.isActive()) { + ch.eventLoop().execute(new PingTask(ch)); + } + } } super.userEventTriggered(ctx, evt); } @@ -69,6 +77,8 @@ protected void channelRead0(ChannelHandlerContext ctx, PongWebSocketFrame msg) t @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + active.set(false); + channel = null; shutdown(); super.channelInactive(ctx); } @@ -84,24 +94,33 @@ private void shutdown() { } private class PingTask implements Runnable { + private final Channel target; + + private PingTask(Channel target) { + this.target = target; + } + @Override public void run() { + if (target == null || !target.isActive()) { + return; + } if (!timeouts.isEmpty()) { return; } final String seq = UUID.randomUUID().toString(); ByteBuf byteBuf = Unpooled.copiedBuffer(seq.getBytes()); PingWebSocketFrame frame = new PingWebSocketFrame(byteBuf); - channel.writeAndFlush(frame).addListener(future -> { + target.writeAndFlush(frame).addListener(future -> { if (future.isSuccess()) { Timeout pingTimeout = TIMER.newTimeout(timeout -> { LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); timeouts.remove(seq); - channel.close(); + target.close(); }, timeout.toMillis(), TimeUnit.MILLISECONDS); timeouts.put(seq, pingTimeout); } else { - channel.close(); + target.close(); } }); } diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java index c3a2dbb..71c7812 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java @@ -53,5 +53,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E } else { LOGGER.error("[DingTalk] connection operation failed, connectionId={}", connectionId, cause); } + // Close the channel so session pool can reconnect instead of leaving a half-dead connection. + if (ctx != null && ctx.channel() != null && ctx.channel().isActive()) { + ctx.close(); + } } } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java new file mode 100644 index 0000000..cce4ed9 --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -0,0 +1,42 @@ +package com.dingtalk.open.app.stream.network.ws; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; +import io.netty.handler.timeout.IdleStateEvent; +import org.junit.Assert; +import org.junit.Test; + +import java.time.Duration; + +/** + * Regression for Aone 84622857 / customer logs: + * IdleStateEvent before WS handshake must not NPE in KeepAliveHandler. + */ +public class KeepAliveHandlerTest { + + @Test + public void idleBeforeHandshakeDoesNotThrow() { + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(handler); + try { + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + Assert.assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void idleAfterHandshakeDoesNotThrow() { + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(handler); + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + Assert.assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } +} From a5f55635bb297c89e19f1376c4a0fd21edca04bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Thu, 23 Jul 2026 15:52:56 +0800 Subject: [PATCH 2/5] test: cover idle-before-handshake and exceptionCaught close Strengthen regression for Aone 84622857 KeepAlive NPE and ensure NettyClientHandler closes channel on connection failures. Co-authored-by: Cursor --- .../network/ws/KeepAliveHandlerTest.java | 3 ++ .../network/ws/NettyClientHandlerTest.java | 40 +++++++++++++++++++ 2 files changed, 43 insertions(+) create mode 100644 app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java index cce4ed9..f444c67 100644 --- a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -19,7 +19,10 @@ public void idleBeforeHandshakeDoesNotThrow() { KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); EmbeddedChannel channel = new EmbeddedChannel(handler); try { + // Multiple idle events before handshake — reproduces Aone 84622857 NPE path. channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.WRITER_IDLE_STATE_EVENT); Assert.assertTrue(channel.isActive()); } finally { channel.finishAndReleaseAll(); diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java new file mode 100644 index 0000000..b93777b --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java @@ -0,0 +1,40 @@ +package com.dingtalk.open.app.stream.network.ws; + +import com.dingtalk.open.app.stream.network.api.ClientConnectionListener; +import com.dingtalk.open.app.stream.network.api.Context; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Ensure exceptionCaught closes the channel so session pool can reconnect. + */ +public class NettyClientHandlerTest { + + @Test + public void exceptionCaughtClosesActiveChannel() { + final AtomicBoolean disconnected = new AtomicBoolean(false); + NettyClientHandler handler = new NettyClientHandler("conn-test", new ClientConnectionListener() { + @Override + public void receive(Context context) { + } + + @Override + public void onDisConnection(String connectionId) { + disconnected.set(true); + } + }); + + EmbeddedChannel channel = new EmbeddedChannel(handler); + Assert.assertTrue(channel.isActive()); + channel.pipeline().fireExceptionCaught(new RuntimeException("simulated io failure")); + + // EmbeddedChannel closes synchronously on close() + Assert.assertFalse("channel should be closed after exceptionCaught", channel.isActive()); + // channelInactive should notify listener + Assert.assertTrue("onDisConnection should be triggered", disconnected.get()); + channel.finishAndReleaseAll(); + } +} From c09c8c90a074d050a536954f461e1f5eada1fce5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Wed, 29 Jul 2026 10:36:23 +0800 Subject: [PATCH 3/5] fix: strengthen websocket error handling tests --- .../open/app/stream/network/ws/NettyClientHandler.java | 4 +--- .../app/stream/network/ws/KeepAliveHandlerTest.java | 10 +++++++++- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java index 71c7812..cd26855 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java @@ -54,8 +54,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E LOGGER.error("[DingTalk] connection operation failed, connectionId={}", connectionId, cause); } // Close the channel so session pool can reconnect instead of leaving a half-dead connection. - if (ctx != null && ctx.channel() != null && ctx.channel().isActive()) { - ctx.close(); - } + ctx.close(); } } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java index f444c67..126080d 100644 --- a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -1,8 +1,10 @@ package com.dingtalk.open.app.stream.network.ws; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; import io.netty.handler.timeout.IdleStateEvent; +import io.netty.util.ReferenceCountUtil; import org.junit.Assert; import org.junit.Test; @@ -30,15 +32,21 @@ public void idleBeforeHandshakeDoesNotThrow() { } @Test - public void idleAfterHandshakeDoesNotThrow() { + public void idleAfterHandshakeSendsPing() { KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); EmbeddedChannel channel = new EmbeddedChannel(handler); + Object outbound = null; try { channel.pipeline().fireUserEventTriggered( WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + channel.runPendingTasks(); + outbound = channel.readOutbound(); + Assert.assertTrue(outbound instanceof PingWebSocketFrame); + Assert.assertNull(channel.readOutbound()); Assert.assertTrue(channel.isActive()); } finally { + ReferenceCountUtil.release(outbound); channel.finishAndReleaseAll(); } } From 72caece357496caf52ad538c3e11a96e84a4efa3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Wed, 29 Jul 2026 22:21:33 +0800 Subject: [PATCH 4/5] fix: close timed-out connections and deduplicate pings --- .../stream/network/ws/KeepAliveHandler.java | 78 +++++++++++------ .../network/ws/ProtocolConnectHandler.java | 31 ++++++- .../network/ws/KeepAliveHandlerTest.java | 68 +++++++++++++++ .../ws/ProtocolConnectHandlerTest.java | 84 +++++++++++++++++++ 4 files changed, 231 insertions(+), 30 deletions(-) create mode 100644 app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java index 5f8b81a..7105d64 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java @@ -15,14 +15,12 @@ import io.netty.util.HashedWheelTimer; import io.netty.util.Timeout; +import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Iterator; -import java.util.Map; -import java.util.Map.Entry; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; /** * @author feiyin @@ -33,13 +31,13 @@ public class KeepAliveHandler extends SimpleChannelInboundHandler timeouts; private final AtomicBoolean active; + private final AtomicReference pendingPing; public KeepAliveHandler(Duration timeout) { this.timeout = timeout; this.active = new AtomicBoolean(false); - this.timeouts = new ConcurrentHashMap<>(); + this.pendingPing = new AtomicReference<>(); } @@ -68,9 +66,11 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc @Override protected void channelRead0(ChannelHandlerContext ctx, PongWebSocketFrame msg) throws Exception { byte[] data = NettyByteBufUtils.getBytes(msg.content()); - Timeout out = timeouts.remove(new String(data)); - if (out != null) { - out.cancel(); + PendingPing ping = pendingPing.get(); + if (ping != null + && ping.seq.equals(new String(data, StandardCharsets.UTF_8)) + && pendingPing.compareAndSet(ping, null)) { + ping.cancelTimeout(); } } @@ -85,11 +85,25 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { private void shutdown() { - Iterator> it = timeouts.entrySet().iterator(); - while (it.hasNext()) { - Entry entry = it.next(); - entry.getValue().cancel(); - it.remove(); + PendingPing ping = pendingPing.getAndSet(null); + if (ping != null) { + ping.cancelTimeout(); + } + } + + private static class PendingPing { + private final String seq; + private volatile Timeout timeout; + + private PendingPing(String seq) { + this.seq = seq; + } + + private void cancelTimeout() { + Timeout current = timeout; + if (current != null) { + current.cancel(); + } } } @@ -102,24 +116,36 @@ private PingTask(Channel target) { @Override public void run() { - if (target == null || !target.isActive()) { + if (!active.get() || target == null || !target.isActive()) { + return; + } + final String seq = UUID.randomUUID().toString(); + final PendingPing ping = new PendingPing(seq); + if (!pendingPing.compareAndSet(null, ping)) { return; } - if (!timeouts.isEmpty()) { + + Timeout pingTimeout = TIMER.newTimeout(ignored -> { + if (pendingPing.compareAndSet(ping, null)) { + active.set(false); + LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); + target.close(); + } + }, timeout.toMillis(), TimeUnit.MILLISECONDS); + ping.timeout = pingTimeout; + + // channelInactive may clear the pending ping while the timeout is being installed. + if (pendingPing.get() != ping) { + pingTimeout.cancel(); return; } - final String seq = UUID.randomUUID().toString(); - ByteBuf byteBuf = Unpooled.copiedBuffer(seq.getBytes()); + + ByteBuf byteBuf = Unpooled.copiedBuffer(seq.getBytes(StandardCharsets.UTF_8)); PingWebSocketFrame frame = new PingWebSocketFrame(byteBuf); target.writeAndFlush(frame).addListener(future -> { - if (future.isSuccess()) { - Timeout pingTimeout = TIMER.newTimeout(timeout -> { - LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); - timeouts.remove(seq); - target.close(); - }, timeout.toMillis(), TimeUnit.MILLISECONDS); - timeouts.put(seq, pingTimeout); - } else { + if (!future.isSuccess() && pendingPing.compareAndSet(ping, null)) { + active.set(false); + pingTimeout.cancel(); target.close(); } }); diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java index 5b7019e..c917e1d 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java @@ -9,6 +9,7 @@ import io.netty.util.Timeout; import java.net.SocketAddress; +import java.nio.channels.ClosedChannelException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -36,10 +37,11 @@ public ProtocolConnectHandler(CompletableFuture future, Long connectTim @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) { - if (timeout != null && !timeout.isExpired()) { - timeout.cancel(); + if (future.complete(ctx.channel())) { + if (timeout != null) { + timeout.cancel(); + } //执行回调 - future.complete(ctx.channel()); ctx.pipeline().remove(ProtocolConnectHandler.class); } } @@ -49,8 +51,29 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc @Override public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception { this.timeout = TIMER.newTimeout(t -> { - ProtocolConnectHandler.this.future.completeExceptionally(new TimeoutException("connect timeout")); + if (ProtocolConnectHandler.this.future.completeExceptionally(new TimeoutException("connect timeout"))) { + ctx.close(); + } }, connectTimeout, TimeUnit.MILLISECONDS); super.connect(ctx, remoteAddress, localAddress, promise); + promise.addListener(connectFuture -> { + if (!connectFuture.isSuccess()) { + Throwable cause = connectFuture.cause() != null + ? connectFuture.cause() : new ClosedChannelException(); + if (future.completeExceptionally(cause)) { + timeout.cancel(); + ctx.close(); + } + } + }); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + future.completeExceptionally(new ClosedChannelException()); + if (timeout != null) { + timeout.cancel(); + } + super.channelInactive(ctx); } } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java index 126080d..fe3978f 100644 --- a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -1,5 +1,8 @@ package com.dingtalk.open.app.stream.network.ws; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; @@ -8,7 +11,11 @@ import org.junit.Assert; import org.junit.Test; +import java.io.IOException; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; /** * Regression for Aone 84622857 / customer logs: @@ -50,4 +57,65 @@ public void idleAfterHandshakeSendsPing() { channel.finishAndReleaseAll(); } } + + @Test + public void multipleIdleEventsQueueOnlyOnePingWhileWriteIsPending() { + DelayedWriteHandler delayedWrite = new DelayedWriteHandler(); + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(delayedWrite, handler); + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.WRITER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.ALL_IDLE_STATE_EVENT); + channel.runPendingTasks(); + + Assert.assertEquals("only one ping may be in flight", 1, delayedWrite.writes.get()); + } finally { + delayedWrite.failPending(); + channel.finishAndReleaseAll(); + } + } + + @Test + public void pendingPingWriteStillTimesOut() throws Exception { + DelayedWriteHandler delayedWrite = new DelayedWriteHandler(); + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofMillis(10)); + EmbeddedChannel channel = new EmbeddedChannel(delayedWrite, handler); + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.runPendingTasks(); + + long deadline = System.nanoTime() + Duration.ofSeconds(2).toNanos(); + while (channel.isActive() && System.nanoTime() < deadline) { + Thread.sleep(10); + channel.runPendingTasks(); + } + Assert.assertFalse("pending ping write must be bounded by keepAlive timeout", channel.isActive()); + } finally { + delayedWrite.failPending(); + channel.finishAndReleaseAll(); + } + } + + private static class DelayedWriteHandler extends ChannelOutboundHandlerAdapter { + private final AtomicInteger writes = new AtomicInteger(); + private final List promises = new ArrayList<>(); + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + writes.incrementAndGet(); + ReferenceCountUtil.release(msg); + promises.add(promise); + } + + private void failPending() { + for (ChannelPromise promise : promises) { + promise.tryFailure(new IOException("test cleanup")); + } + } + } } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java new file mode 100644 index 0000000..8fae145 --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java @@ -0,0 +1,84 @@ +package com.dingtalk.open.app.stream.network.ws; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; +import org.junit.Assert; +import org.junit.Test; + +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +public class ProtocolConnectHandlerTest { + + @Test + public void connectTimeoutClosesChannel() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + EmbeddedChannel channel = new EmbeddedChannel(new ProtocolConnectHandler(future, 10L)); + try { + channel.connect(new InetSocketAddress("localhost", 1234)).syncUninterruptibly(); + try { + future.get(1, TimeUnit.SECONDS); + Assert.fail("connect future should time out"); + } catch (ExecutionException expected) { + // expected + } + channel.runPendingTasks(); + Assert.assertFalse("timed-out connection must be closed", channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void completedHandshakeIsNotClosedByTimeout() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + EmbeddedChannel channel = new EmbeddedChannel(new ProtocolConnectHandler(future, 10L)); + try { + channel.connect(new InetSocketAddress("localhost", 1234)).syncUninterruptibly(); + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + + Assert.assertSame(channel, future.get(1, TimeUnit.SECONDS)); + Thread.sleep(200); + channel.runPendingTasks(); + Assert.assertTrue("completed handshake must win over timeout", channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void connectionFailureCompletesFutureAndClosesChannel() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + ChannelOutboundHandlerAdapter failingConnect = new ChannelOutboundHandlerAdapter() { + @Override + public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, + SocketAddress localAddress, ChannelPromise promise) { + promise.setFailure(new ConnectException("simulated connect failure")); + } + }; + EmbeddedChannel channel = new EmbeddedChannel( + failingConnect, new ProtocolConnectHandler(future, 5000L)); + try { + channel.connect(new InetSocketAddress("localhost", 1234)); + try { + future.get(1, TimeUnit.SECONDS); + Assert.fail("connect future should fail"); + } catch (ExecutionException expected) { + Assert.assertTrue(expected.getCause() instanceof ConnectException); + } + channel.runPendingTasks(); + Assert.assertFalse("failed connection must be closed", channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } +} From 07665b36b15a3824f0477a0502c28db5d5b3d16a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=89=E6=BE=9C?= Date: Thu, 30 Jul 2026 12:16:45 +0800 Subject: [PATCH 5/5] fix: release shared network resources cleanly --- .../app/api/OpenDingTalkStreamClient.java | 44 ++++++++++++++----- .../app-stream-network-api/pom.xml | 5 +++ .../network/api/NetworkSharedResources.java | 25 +++++++++-- .../exception/DingTalkNetworkException.java | 16 +++++++ .../api/NetworkSharedResourcesTest.java | 28 ++++++++++++ .../DingTalkNetworkExceptionTest.java | 27 ++++++++++++ .../app/stream/network/core/Connector.java | 18 ++++++-- .../network/core/DefaultSessionPool.java | 20 +++++---- .../stream/network/ws/KeepAliveHandler.java | 16 +++---- .../network/ws/ProtocolConnectHandler.java | 15 +++---- .../ws/ProtocolConnectHandlerTest.java | 5 +++ 11 files changed, 173 insertions(+), 46 deletions(-) create mode 100644 app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java create mode 100644 app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java diff --git a/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java b/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java index 13c2309..fcdfded 100644 --- a/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java +++ b/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java @@ -11,6 +11,7 @@ import com.dingtalk.open.app.stream.network.api.ClientConnectionListener; import com.dingtalk.open.app.stream.network.api.EndPointConnection; import com.dingtalk.open.app.stream.network.api.NetProxy; +import com.dingtalk.open.app.stream.network.api.NetworkSharedResources; import com.dingtalk.open.app.stream.network.core.EndPointConnectionFactory; import com.dingtalk.open.app.stream.network.core.NetWorkService; import com.dingtalk.open.app.stream.network.core.Subscription; @@ -35,6 +36,7 @@ class OpenDingTalkStreamClient implements OpenDingTalkClient { private NetWorkService netWorkService; private OpenApiClient openApiClient; private Set subscriptions; + private boolean networkResourcesAcquired; public OpenDingTalkStreamClient(DingTalkCredential credential, CommandDispatcher dispatcher, ExecutorService executor, ClientOption option, Set subscriptions, NetProxy netProxy) { @@ -50,12 +52,19 @@ public OpenDingTalkStreamClient(DingTalkCredential credential, CommandDispatcher @Override public synchronized void start() throws OpenDingTalkAppException { if (status.get() == Status.INIT) { - this.openApiClient = OpenApiClientBuilder.create().setHost(option.getOpenApiHost()).setTimeout(option.getConnectionTTL()).setProxy(netProxy).build(); - final EndPointConnectionFactory factory = () -> openConnection(this.credential, subscriptions, netProxy); - ClientConnectionListener listener = new AppServiceListener(dispatcher, executor); - this.netWorkService = new NetWorkService(factory, listener, option.getMaxConnectionCount(), option.getConnectionTTL(), option.getConnectTimeout(), option.getKeepAliveOption().getKeepAliveIdleMill()); - this.netWorkService.start(); - this.status.set(Status.ACTIVE); + NetworkSharedResources.acquireNetWorkEventLoopGroup(); + networkResourcesAcquired = true; + try { + this.openApiClient = OpenApiClientBuilder.create().setHost(option.getOpenApiHost()).setTimeout(option.getConnectionTTL()).setProxy(netProxy).build(); + final EndPointConnectionFactory factory = () -> openConnection(this.credential, subscriptions, netProxy); + ClientConnectionListener listener = new AppServiceListener(dispatcher, executor); + this.netWorkService = new NetWorkService(factory, listener, option.getMaxConnectionCount(), option.getConnectionTTL(), option.getConnectTimeout(), option.getKeepAliveOption().getKeepAliveIdleMill()); + this.netWorkService.start(); + this.status.set(Status.ACTIVE); + } catch (RuntimeException | Error e) { + releaseNetworkResources(); + throw e; + } } else if (status.get() == Status.INACTIVE) { throw new OpenDingTalkAppException(DingTalkAppError.CLIENT_STATE_ERROR); } @@ -64,13 +73,24 @@ public synchronized void start() throws OpenDingTalkAppException { @Override public synchronized void stop() throws Exception { if (status.get() == Status.ACTIVE) { - if (this.netWorkService != null) { - this.netWorkService.shutdown(); - } - if (executor != null) { - this.executor.shutdown(); + try { + if (this.netWorkService != null) { + this.netWorkService.shutdown(); + } + if (executor != null) { + this.executor.shutdown(); + } + } finally { + releaseNetworkResources(); + status.set(Status.INACTIVE); } - status.set(Status.INACTIVE); + } + } + + private void releaseNetworkResources() { + if (networkResourcesAcquired) { + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + networkResourcesAcquired = false; } } diff --git a/app-stream-network/app-stream-network-api/pom.xml b/app-stream-network/app-stream-network-api/pom.xml index faa0c84..600e18c 100644 --- a/app-stream-network/app-stream-network-api/pom.xml +++ b/app-stream-network/app-stream-network-api/pom.xml @@ -15,6 +15,11 @@ app-stream-network-api + + junit + junit + test + com.dingtalk.open app-stream-protocol diff --git a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java index 219dbed..26a1971 100644 --- a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java +++ b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java @@ -11,14 +11,31 @@ public class NetworkSharedResources { private static NioEventLoopGroup EVENT_LOOP_GROUP; + private static int referenceCount; - static { - EVENT_LOOP_GROUP = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2); + public static synchronized EventLoopGroup acquireNetWorkEventLoopGroup() { + referenceCount++; + return getNetWorkEventLoopGroup(); } - public static EventLoopGroup getNetWorkEventLoopGroup() { + public static synchronized EventLoopGroup getNetWorkEventLoopGroup() { + if (EVENT_LOOP_GROUP == null + || EVENT_LOOP_GROUP.isShuttingDown() + || EVENT_LOOP_GROUP.isShutdown() + || EVENT_LOOP_GROUP.isTerminated()) { + EVENT_LOOP_GROUP = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2); + } return EVENT_LOOP_GROUP; } - + public static synchronized void releaseNetWorkEventLoopGroup() { + if (referenceCount == 0) { + return; + } + referenceCount--; + if (referenceCount == 0 && EVENT_LOOP_GROUP != null) { + EVENT_LOOP_GROUP.shutdownGracefully(); + EVENT_LOOP_GROUP = null; + } + } } diff --git a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java index deb3929..98c8d86 100644 --- a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java +++ b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java @@ -6,9 +6,25 @@ */ public class DingTalkNetworkException extends RuntimeException { + private final NetWorkError netWorkError; public DingTalkNetworkException(NetWorkError netWorkError) { + super(netWorkError != null ? netWorkError.name() : null); + this.netWorkError = netWorkError; + } + + public DingTalkNetworkException(NetWorkError netWorkError, Throwable cause) { + super(netWorkError != null ? netWorkError.name() : null, cause); + this.netWorkError = netWorkError; + } + + public DingTalkNetworkException(NetWorkError netWorkError, String detail) { + super(netWorkError != null ? netWorkError.name() + ": " + detail : detail); + this.netWorkError = netWorkError; + } + public NetWorkError getNetWorkError() { + return netWorkError; } } diff --git a/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java new file mode 100644 index 0000000..e57821f --- /dev/null +++ b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java @@ -0,0 +1,28 @@ +package com.dingtalk.open.app.stream.network.api; + +import io.netty.channel.EventLoopGroup; +import org.junit.Assert; +import org.junit.Test; + +public class NetworkSharedResourcesTest { + + @Test + public void sharedEventLoopStopsAfterLastRelease() { + EventLoopGroup first = NetworkSharedResources.acquireNetWorkEventLoopGroup(); + EventLoopGroup second = NetworkSharedResources.acquireNetWorkEventLoopGroup(); + Assert.assertSame(first, second); + + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + Assert.assertFalse("one active client still owns the event loop", first.isShuttingDown()); + + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + Assert.assertTrue("last client release must stop the event loop", first.isShuttingDown()); + + EventLoopGroup replacement = NetworkSharedResources.acquireNetWorkEventLoopGroup(); + try { + Assert.assertNotSame("a stopped event loop must not be reused", first, replacement); + } finally { + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + } + } +} diff --git a/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java new file mode 100644 index 0000000..438f3c3 --- /dev/null +++ b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java @@ -0,0 +1,27 @@ +package com.dingtalk.open.app.stream.network.api.exception; + +import org.junit.Assert; +import org.junit.Test; + +public class DingTalkNetworkExceptionTest { + + @Test + public void preservesNetworkErrorAndCause() { + IllegalStateException cause = new IllegalStateException("service loader failed"); + DingTalkNetworkException exception = + new DingTalkNetworkException(NetWorkError.OPEN_CONNECTION_ERROR, cause); + + Assert.assertEquals(NetWorkError.OPEN_CONNECTION_ERROR, exception.getNetWorkError()); + Assert.assertEquals("OPEN_CONNECTION_ERROR", exception.getMessage()); + Assert.assertSame(cause, exception.getCause()); + } + + @Test + public void includesProtocolDetail() { + DingTalkNetworkException exception = + new DingTalkNetworkException(NetWorkError.PROTOCOL_ILLEGAL, "protocol WSS"); + + Assert.assertEquals(NetWorkError.PROTOCOL_ILLEGAL, exception.getNetWorkError()); + Assert.assertEquals("PROTOCOL_ILLEGAL: protocol WSS", exception.getMessage()); + } +} diff --git a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java index 847ab96..6eb3a76 100644 --- a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java +++ b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java @@ -9,6 +9,7 @@ import java.time.Duration; import java.util.Iterator; import java.util.Map; +import java.util.ServiceConfigurationError; import java.util.ServiceLoader; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.Lock; @@ -39,7 +40,9 @@ public static Session connect(EndPointConnection connection, ClientConnectionLis ensureActive(); TransportConnector connector = CONNECTOR_REGISTRY.get(connection.getProtocol()); if (connector == null) { - throw new DingTalkNetworkException(NetWorkError.PROTOCOL_ILLEGAL); + throw new DingTalkNetworkException( + NetWorkError.PROTOCOL_ILLEGAL, + "no transport connector registered for protocol " + connection.getProtocol()); } ConnectOption option = ConnectOption.builder().setTimeout(timeout).setTtl(ttl).setKeepAliveIdle(Duration.ofMillis(keepAliveIdle)).setKeepAliveTimeout(KEEP_ALIVE_TIMEOUT).build(); return connector.connect(connection, listener, option); @@ -55,7 +58,8 @@ private static void ensureActive() { if (INIT) { return; } - ServiceLoader transportConnectors = ServiceLoader.load(TransportConnector.class); + ServiceLoader transportConnectors = ServiceLoader.load( + TransportConnector.class, Connector.class.getClassLoader()); Iterator it = transportConnectors.iterator(); while (it.hasNext()) { TransportConnector transportConnector = it.next(); @@ -67,9 +71,17 @@ private static void ensureActive() { } } } + if (CONNECTOR_REGISTRY.isEmpty()) { + throw new DingTalkNetworkException( + NetWorkError.OPEN_CONNECTION_ERROR, + "no TransportConnector implementation was found by ServiceLoader"); + } INIT = true; - } catch (Exception e) { + } catch (DingTalkNetworkException e) { + throw e; + } catch (ServiceConfigurationError | RuntimeException e) { LOGGER.error("[DingTalk] client init transport failed, {}", e); + throw new DingTalkNetworkException(NetWorkError.OPEN_CONNECTION_ERROR, e); } finally { INIT_LOCK.unlock(); } diff --git a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java index 0bf97ad..c8da9cc 100644 --- a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java +++ b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java @@ -97,16 +97,14 @@ private void evict() { @Override public void shutdown() { if (status.compareAndSet(true, false)) { - scheduledExecutorService.execute(() -> { - for (String sessionId : sessions.keySet()) { - try { - closeSession(sessionId); - } catch (Exception e) { - LOGGER.error("[DingTalk] close session failed, connectionId={}", sessionId, e); - } + scheduledExecutorService.shutdownNow(); + for (String sessionId : sessions.keySet()) { + try { + closeSession(sessionId); + } catch (Exception e) { + LOGGER.error("[DingTalk] close session failed, connectionId={}", sessionId, e); } - }); - scheduledExecutorService.execute(scheduledExecutorService::shutdown); + } } } @@ -169,6 +167,10 @@ public void run() { if (session == null) { return; } + if (!isActive()) { + session.close(); + return; + } LOGGER.info("[DingTalk] connection is established, connectionId={}", session.getId()); previous = sessions.put(connection.getConnectionId(), session); if (previous != null) { diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java index 7105d64..3344472 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java @@ -12,12 +12,11 @@ import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; import io.netty.handler.timeout.IdleStateEvent; -import io.netty.util.HashedWheelTimer; -import io.netty.util.Timeout; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.UUID; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -29,7 +28,6 @@ public class KeepAliveHandler extends SimpleChannelInboundHandler { private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(KeepAliveHandler.class); private final Duration timeout; - private final static HashedWheelTimer TIMER = new HashedWheelTimer(); private volatile Channel channel; private final AtomicBoolean active; private final AtomicReference pendingPing; @@ -93,16 +91,16 @@ private void shutdown() { private static class PendingPing { private final String seq; - private volatile Timeout timeout; + private volatile ScheduledFuture timeout; private PendingPing(String seq) { this.seq = seq; } private void cancelTimeout() { - Timeout current = timeout; + ScheduledFuture current = timeout; if (current != null) { - current.cancel(); + current.cancel(false); } } } @@ -125,7 +123,7 @@ public void run() { return; } - Timeout pingTimeout = TIMER.newTimeout(ignored -> { + ScheduledFuture pingTimeout = target.eventLoop().schedule(() -> { if (pendingPing.compareAndSet(ping, null)) { active.set(false); LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); @@ -136,7 +134,7 @@ public void run() { // channelInactive may clear the pending ping while the timeout is being installed. if (pendingPing.get() != ping) { - pingTimeout.cancel(); + pingTimeout.cancel(false); return; } @@ -145,7 +143,7 @@ public void run() { target.writeAndFlush(frame).addListener(future -> { if (!future.isSuccess() && pendingPing.compareAndSet(ping, null)) { active.set(false); - pingTimeout.cancel(); + pingTimeout.cancel(false); target.close(); } }); diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java index c917e1d..a719697 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java @@ -5,12 +5,11 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; -import io.netty.util.HashedWheelTimer; -import io.netty.util.Timeout; import java.net.SocketAddress; import java.nio.channels.ClosedChannelException; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -19,9 +18,7 @@ * @date 2023/9/7 */ public class ProtocolConnectHandler extends ChannelDuplexHandler { - private static final HashedWheelTimer TIMER = new HashedWheelTimer(); - - private Timeout timeout; + private ScheduledFuture timeout; private final CompletableFuture future; /** @@ -39,7 +36,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) { if (future.complete(ctx.channel())) { if (timeout != null) { - timeout.cancel(); + timeout.cancel(false); } //执行回调 ctx.pipeline().remove(ProtocolConnectHandler.class); @@ -50,7 +47,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc @Override public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception { - this.timeout = TIMER.newTimeout(t -> { + this.timeout = ctx.executor().schedule(() -> { if (ProtocolConnectHandler.this.future.completeExceptionally(new TimeoutException("connect timeout"))) { ctx.close(); } @@ -61,7 +58,7 @@ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, Sock Throwable cause = connectFuture.cause() != null ? connectFuture.cause() : new ClosedChannelException(); if (future.completeExceptionally(cause)) { - timeout.cancel(); + timeout.cancel(false); ctx.close(); } } @@ -72,7 +69,7 @@ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, Sock public void channelInactive(ChannelHandlerContext ctx) throws Exception { future.completeExceptionally(new ClosedChannelException()); if (timeout != null) { - timeout.cancel(); + timeout.cancel(false); } super.channelInactive(ctx); } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java index 8fae145..d628d8d 100644 --- a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java @@ -24,6 +24,11 @@ public void connectTimeoutClosesChannel() throws Exception { EmbeddedChannel channel = new EmbeddedChannel(new ProtocolConnectHandler(future, 10L)); try { channel.connect(new InetSocketAddress("localhost", 1234)).syncUninterruptibly(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (!future.isDone() && System.nanoTime() < deadline) { + Thread.sleep(10); + channel.runScheduledPendingTasks(); + } try { future.get(1, TimeUnit.SECONDS); Assert.fail("connect future should time out");