### Question:
ChannelFlowManager { public void notifyMessageAddition(ChannelHandlerContext ctx) { messagesInFlight++; if (messagesInFlight > highLimit && inflowEnabled) { inflowEnabled = false; ctx.writeAndFlush(new ChannelFlow(channel.getChannelId(), false)); ctx.channel().config().setAutoRead(false); LOGGER.info("Inflow disabled for channel {}-{}", channel.getChannelId(), ctx.channel().remoteAddress()); } } ChannelFlowManager(AmqpChannel channel, int lowLimit, int highLimit); void notifyMessageAddition(ChannelHandlerContext ctx); void notifyMessageRemoval(ChannelHandlerContext ctx);  }### Answer:
@Test public void testFlowNotDisabledWhenHighLevelNotExceeded() throws Exception { IntStream.rangeClosed(1, 10) .forEach(i -> channelFlowManager.notifyMessageAddition(ctx)); Mockito.verify(ctx, Mockito.never()).writeAndFlush(argumentCaptor.capture()); }
@Test public void testFlowDisabledWhenHighLevelExceeded() throws Exception { IntStream.rangeClosed(1, 11) .forEach(i -> channelFlowManager.notifyMessageAddition(ctx)); Mockito.verify(ctx, Mockito.times(1)).writeAndFlush(argumentCaptor.capture()); }