@Test void failConnectIfPromiseTimesOut() throws Exception { Duration timeout = Duration.ofMillis(100); when(timeoutConfig.connectTimeout()).thenReturn(timeout); FeatureNegotiatingHandler handler = new FeatureNegotiatingHandler( endpointContext, Collections.singleton(ServerFeature.TRACING) ); channel.pipeline().addLast(handler); final ChannelFuture connect = channel.connect( new InetSocketAddress("1.2.3.4", 1234) ); channel.pipeline().fireChannelActive(); Thread.sleep(timeout.toMillis() + 5); channel.runScheduledPendingTasks(); assertTrue(connect.isDone()); assertTrue(connect.cause() instanceof TimeoutException); assertEquals("KV Feature Negotiation timed out after 100ms", connect.cause().getMessage()); }
@Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { if (features.isEmpty()) { ConnectTimings.record(ctx.channel(), this.getClass()); ctx.pipeline().remove(this); ctx.connect(remoteAddress, localAddress, promise); } else { interceptedConnectPromise = promise; ChannelPromise downstream = ctx.newPromise(); downstream.addListener(f -> { if (!f.isSuccess() && !interceptedConnectPromise.isDone()) { ConnectTimings.record(ctx.channel(), this.getClass()); interceptedConnectPromise.tryFailure(f.cause()); } }); ctx.connect(remoteAddress, localAddress, downstream); } }
FeatureNegotiatingHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { if (features.isEmpty()) { ConnectTimings.record(ctx.channel(), this.getClass()); ctx.pipeline().remove(this); ctx.connect(remoteAddress, localAddress, promise); } else { interceptedConnectPromise = promise; ChannelPromise downstream = ctx.newPromise(); downstream.addListener(f -> { if (!f.isSuccess() && !interceptedConnectPromise.isDone()) { ConnectTimings.record(ctx.channel(), this.getClass()); interceptedConnectPromise.tryFailure(f.cause()); } }); ctx.connect(remoteAddress, localAddress, downstream); } } }
FeatureNegotiatingHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { if (features.isEmpty()) { ConnectTimings.record(ctx.channel(), this.getClass()); ctx.pipeline().remove(this); ctx.connect(remoteAddress, localAddress, promise); } else { interceptedConnectPromise = promise; ChannelPromise downstream = ctx.newPromise(); downstream.addListener(f -> { if (!f.isSuccess() && !interceptedConnectPromise.isDone()) { ConnectTimings.record(ctx.channel(), this.getClass()); interceptedConnectPromise.tryFailure(f.cause()); } }); ctx.connect(remoteAddress, localAddress, downstream); } } FeatureNegotiatingHandler(final EndpointContext endpointContext,
                            final Set<ServerFeature> features); }
FeatureNegotiatingHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { if (features.isEmpty()) { ConnectTimings.record(ctx.channel(), this.getClass()); ctx.pipeline().remove(this); ctx.connect(remoteAddress, localAddress, promise); } else { interceptedConnectPromise = promise; ChannelPromise downstream = ctx.newPromise(); downstream.addListener(f -> { if (!f.isSuccess() && !interceptedConnectPromise.isDone()) { ConnectTimings.record(ctx.channel(), this.getClass()); interceptedConnectPromise.tryFailure(f.cause()); } }); ctx.connect(remoteAddress, localAddress, downstream); } } FeatureNegotiatingHandler(final EndpointContext endpointContext,
                            final Set<ServerFeature> features); @Override void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress,
                      final SocketAddress localAddress, final ChannelPromise promise); @Override void channelActive(final ChannelHandlerContext ctx); @Override void channelRead(final ChannelHandlerContext ctx, final Object msg); @Override void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause); }
FeatureNegotiatingHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { if (features.isEmpty()) { ConnectTimings.record(ctx.channel(), this.getClass()); ctx.pipeline().remove(this); ctx.connect(remoteAddress, localAddress, promise); } else { interceptedConnectPromise = promise; ChannelPromise downstream = ctx.newPromise(); downstream.addListener(f -> { if (!f.isSuccess() && !interceptedConnectPromise.isDone()) { ConnectTimings.record(ctx.channel(), this.getClass()); interceptedConnectPromise.tryFailure(f.cause()); } }); ctx.connect(remoteAddress, localAddress, downstream); } } FeatureNegotiatingHandler(final EndpointContext endpointContext,
                            final Set<ServerFeature> features); @Override void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress,
                      final SocketAddress localAddress, final ChannelPromise promise); @Override void channelActive(final ChannelHandlerContext ctx); @Override void channelRead(final ChannelHandlerContext ctx, final Object msg); @Override void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause);  }