@Test void decodeAndIgnoreNonAskedForFeaturesInResponse() { Set<ServerFeature> toNegotiate = EnumSet.of(ServerFeature.SNAPPY, ServerFeature.TRACING); FeatureNegotiatingHandler handler = new FeatureNegotiatingHandler( endpointContext, toNegotiate ); channel.pipeline().addLast(handler); assertEquals(handler, channel.pipeline().get(FeatureNegotiatingHandler.class)); ChannelFuture connectFuture = channel.connect(new InetSocketAddress("1.2.3.4", 1234)); assertFalse(connectFuture.isDone()); channel.pipeline().fireChannelActive(); channel.runPendingTasks(); ByteBuf writtenRequest = channel.readOutbound(); verifyRequest(writtenRequest, MemcacheProtocol.Opcode.HELLO.opcode(), true, false, true); assertNotNull(channel.pipeline().get(FeatureNegotiatingHandler.class)); ByteBuf response = decodeHexDump(readResource( "success_hello_response.txt", FeatureNegotiatingHandlerTest.class )); channel.writeInbound(response); channel.runPendingTasks(); assertTrue(connectFuture.isSuccess()); assertEquals(2, eventBus.publishedEvents().size()); UnsolicitedFeaturesReturnedEvent unsolicitedEvent = (UnsolicitedFeaturesReturnedEvent) eventBus.publishedEvents().get(0); assertEquals( "Received unsolicited features during HELLO [TCPNODELAY, XATTR, XERROR, SELECT_BUCKET]", unsolicitedEvent.description() ); assertEquals(Event.Severity.WARN, unsolicitedEvent.severity()); FeaturesNegotiatedEvent event = (FeaturesNegotiatedEvent) eventBus.publishedEvents().get(1); assertEquals( "Negotiated [SNAPPY, TRACING]", event.description() ); List<ServerFeature> serverFeatures = channel.attr(ChannelAttributes.SERVER_FEATURE_KEY).get(); assertEquals(toNegotiate, new HashSet<>(serverFeatures)); assertNull(channel.pipeline().get(FeatureNegotiatingHandler.class)); ReferenceCountUtil.release(writtenRequest); }
@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);  }