AbstractSessionNettyHandler extends AbstractNettyHandler { @Override public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { logger.debug("[channelWritabilityChanged] writable: {}", ctx.channel().isWritable()); logger.info("[channelWritabilityChanged] buffer size: {}", ctx.channel().unsafe().outboundBuffer().totalPendingWriteBytes()); if(ctx.channel().isWritable()) { session.setWritableState(Session.SessionWritableState.WRITABLE); } else { session.setWritableState(Session.SessionWritableState.UNWRITABLE); } super.channelWritabilityChanged(ctx); }  @Override void channelInactive(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); @Override void channelWritabilityChanged(ChannelHandlerContext ctx);  }
@Test public void channelWritabilityChanged() { EmbeddedChannel channel = new EmbeddedChannel(handler); channel.config().setWriteBufferHighWaterMark(32768 + 200); channel.config().setWriteBufferLowWaterMark(32768); int count = 0; while(count < 32768 + 200) { channel.write(Unpooled.copiedBuffer("hello world!".getBytes())); count += "hello world!".getBytes().length; } Assert.assertFalse(channel.isWritable()); verify(session).setWritableState(Session.SessionWritableState.UNWRITABLE); }