### Question:
AbstractConnectionToChannelBridge extends BackpressureManagingHandler { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof EmitConnectionEvent) { if (!connectionEmitted) { emitNewConnection(ctx.channel()); connectionEmitted = true; } } else if (evt instanceof ConnectionCreationFailedEvent) { if (isValidToEmit(newChannelSub)) { newChannelSub.onError(((ConnectionCreationFailedEvent)evt).getThrowable()); } } else if (evt instanceof ChannelSubscriberEvent) { @SuppressWarnings("unchecked") final ChannelSubscriberEvent<R, W> channelSubscriberEvent = (ChannelSubscriberEvent<R, W>) evt; newConnectionSubscriber(channelSubscriberEvent); } else if (evt instanceof ConnectionInputSubscriberEvent) { @SuppressWarnings("unchecked") ConnectionInputSubscriberEvent<R, W> event = (ConnectionInputSubscriberEvent<R, W>) evt; newConnectionInputSubscriber(ctx.channel(), event.getSubscriber(), false); } else if (evt instanceof ConnectionInputSubscriberResetEvent) { resetConnectionInputSubscriber(); } else if (evt instanceof ConnectionInputSubscriberReplaceEvent) { @SuppressWarnings("unchecked") ConnectionInputSubscriberReplaceEvent<R, W> event = (ConnectionInputSubscriberReplaceEvent<R, W>) evt; replaceConnectionInputSubscriber(ctx.channel(), event); } super.userEventTriggered(ctx, evt); } protected  AbstractConnectionToChannelBridge(String thisHandlerName, ConnectionEventListener eventListener,
                                                EventPublisher eventPublisher); protected  AbstractConnectionToChannelBridge(String thisHandlerName,
                                                AttributeKey<ConnectionEventListener> eventListenerAttributeKey,
                                                AttributeKey<EventPublisher> eventPublisherAttributeKey); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void channelUnregistered(ChannelHandlerContext ctx); @Override void userEventTriggered(ChannelHandlerContext ctx, Object evt); @SuppressWarnings("unchecked") @Override void newMessage(ChannelHandlerContext ctx, Object msg); @Override boolean shouldReadMore(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause);  }### Answer:
@Test(timeout = 60000) public void testInputSubscriptionReset() throws Exception { connectionHandlerRule.channel.config().setAutoRead(false); ConnectionSubscriber subscriber = connectionHandlerRule.enableConnectionSubscriberAndAssert(true); connectionHandlerRule.activateConnectionAndAssert(subscriber); ConnectionInputSubscriber inputSubscriber = connectionHandlerRule.enableConnectionInputSubscriber(); inputSubscriber.assertTerminalEvent(); assertThat("Unexpected on next events for second subscriber.", inputSubscriber.getOnNextEvents(), hasSize(0)); connectionHandlerRule.handler.userEventTriggered(connectionHandlerRule.ctx, new ConnectionInputSubscriberResetEvent() { }); inputSubscriber = connectionHandlerRule.enableConnectionInputSubscriber(); assertThat("Unexpected on next count for input subscriber post reset.", inputSubscriber.getOnNextEvents(), hasSize(0)); assertThat("Unexpected on error count for input subscriber post reset.", inputSubscriber.getOnErrorEvents(), hasSize(0)); assertThat("Unexpected on completed count for input subscriber post reset.", inputSubscriber.getOnCompletedEvents(), hasSize(0)); }