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..f444c67 --- /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,45 @@ +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 { + // 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(); + } + } + + @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(); + } + } +} 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(); + } +}