@Test void failConnectIfPromiseTimesOut() throws Exception { channel = new EmbeddedChannel(); eventBus = new SimpleEventBus(true); CoreEnvironment env = mock(CoreEnvironment.class); TimeoutConfig timeoutConfig = mock(TimeoutConfig.class); when(env.eventBus()).thenReturn(eventBus); when(env.timeoutConfig()).thenReturn(timeoutConfig); when(timeoutConfig.connectTimeout()).thenReturn(Duration.ofMillis(100)); CoreContext coreContext = new CoreContext(mock(Core.class), 1, env, mock(Authenticator.class)); EndpointContext endpointContext = new EndpointContext(coreContext, new HostAndPort("127.0.0.1", 1234), null, ServiceType.KV, Optional.empty(), Optional.empty(), Optional.empty()); SaslListMechanismsHandler handler = new SaslListMechanismsHandler(endpointContext); channel.pipeline().addLast(handler); final ChannelFuture connect = channel.connect( new InetSocketAddress("1.2.3.4", 1234) ); channel.pipeline().fireChannelActive(); Thread.sleep(Duration.ofMillis(100).toMillis() + 5); channel.runScheduledPendingTasks(); assertTrue(connect.isDone()); assertTrue(connect.cause() instanceof TimeoutException); assertEquals("SASL Mechanism listing timed out after 100ms", connect.cause().getMessage()); }
@Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { 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); }
SaslListMechanismsHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { 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); } }
SaslListMechanismsHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { 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); } SaslListMechanismsHandler(final EndpointContext endpointContext); }
SaslListMechanismsHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { 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); } SaslListMechanismsHandler(final EndpointContext endpointContext); @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); }
SaslListMechanismsHandler extends ChannelDuplexHandler { @Override public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) { 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); } SaslListMechanismsHandler(final EndpointContext endpointContext); @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);  }