OvsdbConnectionHandler extends ChannelInboundHandlerAdapter { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { ++readIdleCount; LOGGER.warn( "Read idle of {} for {} times", ctx.channel(), readIdleCount); if (readIdleCount >= MAX_READ_IDLE_BEFORE_CLOSE) { LOGGER.warn("Read idle time reaches the threshold. Closing the channel to {}", ctx.channel()); ctx.close(); } else { String heartbeatMsg = getHeartbeatMsg(); LOGGER.debug("Sending heartbeat {} to channel {}", heartbeatMsg, ctx.channel()); ctx.writeAndFlush(heartbeatMsg); } } else { ctx.fireUserEventTriggered(evt); } } OvsdbConnectionHandler(
      ConnectionCallback connectionCallback, ScheduledExecutorService executorService
  );   OvsdbConnectionHandler(
      CompletableFuture<OvsdbClient> ovsdbClientFuture, ScheduledExecutorService executorService
  ); private  OvsdbConnectionHandler(
      CompletableFuture<OvsdbClient> ovsdbClientFuture, ConnectionCallback connectionCallback,
      ScheduledExecutorService executorService
  ); @Override void channelActive(final ChannelHandlerContext ctx); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void channelInactive(ChannelHandlerContext ctx); @Override void userEventTriggered(ChannelHandlerContext ctx, Object evt);  }
@Test public void testReadRegularEvent() { OvsdbConnectionHandler heartBeatHandler = newOvsdbConnectionHandler(); Object userEvent = new Object(); heartBeatHandler.userEventTriggered(ctx, userEvent); verify(ctx).fireUserEventTriggered(userEvent); }
@Test public void testChannelReadTimeout() { OvsdbConnectionHandler heartBeatHandler = newOvsdbConnectionHandler(); for (int i = 0; i < MAX_READ_IDLE_BEFORE_CLOSE; i++) { heartBeatHandler.userEventTriggered(ctx, idleStateEvent); } verify(ctx).close(); }