ResponseRouter extends SimpleChannelInboundHandler<ResponseMessage> { protected boolean send(@NotNull OutRequest<?> outRequest) { Request request = outRequest.getRequest(); this.requestMap.put(request.getId(), outRequest); setTimer(outRequest); if (!ctx.channel().isActive()) { handleResponse(new ErrorResponse( outRequest.getRequest().getId(), ErrorType.NetworkError, "Channel is closed")); return false; } requestQueue.offer(outRequest.getRequest().getId()); sendRequest(); return true; } ResponseRouter(@NotNull TChannel topChannel, @NotNull HashedWheelTimer timer); @VisibleForTesting  ResponseRouter(@NotNull TChannel topChannel, @NotNull PeerManager peerManager, @NotNull HashedWheelTimer timer); @Override void channelActive(@NotNull ChannelHandlerContext ctx); @Override void channelWritabilityChanged(ChannelHandlerContext ctx); boolean expectResponse(@NotNull OutRequest<?> outRequest); void clean();  }
@Test public void resetConnectionOnContinuousTimeouts() throws Exception { OutRequest or1 = createOutputRequest(1, 100); OutRequest or2 = createOutputRequest(2, 100); PeerManager peerManager = mock(PeerManager.class); ResponseRouter responseRouter = getResponseRouter(peerManager); responseRouter.send(or1); responseRouter.send(or2); Thread.sleep(200); verify(peerManager, times(1)) .handleConnectionErrors(any(Channel.class), any(TChannelConnectionReset.class)); }