class 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());
        }
    }

    public  ChannelFlowManager(AmqpChannel channel, int lowLimit, int highLimit);

}