@Test public void childQueueIsDrainedAndNewDataIsDispatchedInParentReadLoopNoAutoRead() { AtomicInteger numReads = new AtomicInteger(1); final AtomicInteger channelReadCompleteCount = new AtomicInteger(0); final AtomicBoolean shouldDisableAutoRead = new AtomicBoolean(); Consumer<ChannelHandlerContext> ctxConsumer = new Consumer<ChannelHandlerContext>() { @Override public void accept(ChannelHandlerContext obj) { channelReadCompleteCount.incrementAndGet(); if (shouldDisableAutoRead.get()) { obj.channel().config().setAutoRead(false); } } }; LastInboundHandler inboundHandler = streamActiveAndWriteHeaders(inboundStream, numReads, ctxConsumer); Http2StreamChannel childChannel = (Http2StreamChannel) inboundHandler.channel(); childChannel.config().setAutoRead(false); Http2DataFrame dataFrame1 = new DefaultHttp2DataFrame(bb("1")).stream(inboundStream); Http2DataFrame dataFrame2 = new DefaultHttp2DataFrame(bb("2")).stream(inboundStream); Http2DataFrame dataFrame3 = new DefaultHttp2DataFrame(bb("3")).stream(inboundStream); Http2DataFrame dataFrame4 = new DefaultHttp2DataFrame(bb("4")).stream(inboundStream); assertEquals(new DefaultHttp2HeadersFrame(request).stream(inboundStream), inboundHandler.readInbound()); parentChannel.writeOneInbound(new Object()); codec.onHttp2Frame(dataFrame1); assertEquals(dataFrame1, inboundHandler.readInbound()); codec.onHttp2Frame(dataFrame2); numReads.set(2); childChannel.read(); assertEquals(dataFrame2, inboundHandler.readInbound()); assertNull(inboundHandler.readInbound()); codec.onHttp2Frame(dataFrame3); assertEquals(dataFrame3, inboundHandler.readInbound()); codec.onHttp2Frame(dataFrame4); assertNull(inboundHandler.readInbound()); childChannel.read(); assertEquals(dataFrame4, inboundHandler.readInbound()); assertNull(inboundHandler.readInbound()); parentChannel.flushInbound(); assertEquals(4, channelReadCompleteCount.get()); dataFrame1.release(); dataFrame2.release(); dataFrame3.release(); dataFrame4.release(); }
@Override final void onHttp2Frame(ChannelHandlerContext ctx, Http2Frame frame) { if (frame instanceof Http2StreamFrame) { Http2StreamFrame streamFrame = (Http2StreamFrame) frame; ((Http2MultiplexCodecStream) streamFrame.stream()).channel.fireChildRead(streamFrame); } else if (frame instanceof Http2GoAwayFrame) { onHttp2GoAwayFrame(ctx, (Http2GoAwayFrame) frame); ctx.fireChannelRead(frame); } else if (frame instanceof Http2SettingsFrame) { Http2Settings settings = ((Http2SettingsFrame) frame).settings(); if (settings.initialWindowSize() != null) { initialOutboundStreamWindow = settings.initialWindowSize(); } ctx.fireChannelRead(frame); } else { ctx.fireChannelRead(frame); } }
Http2MultiplexCodec extends Http2FrameCodec { @Override final void onHttp2Frame(ChannelHandlerContext ctx, Http2Frame frame) { if (frame instanceof Http2StreamFrame) { Http2StreamFrame streamFrame = (Http2StreamFrame) frame; ((Http2MultiplexCodecStream) streamFrame.stream()).channel.fireChildRead(streamFrame); } else if (frame instanceof Http2GoAwayFrame) { onHttp2GoAwayFrame(ctx, (Http2GoAwayFrame) frame); ctx.fireChannelRead(frame); } else if (frame instanceof Http2SettingsFrame) { Http2Settings settings = ((Http2SettingsFrame) frame).settings(); if (settings.initialWindowSize() != null) { initialOutboundStreamWindow = settings.initialWindowSize(); } ctx.fireChannelRead(frame); } else { ctx.fireChannelRead(frame); } } }
Http2MultiplexCodec extends Http2FrameCodec { @Override final void onHttp2Frame(ChannelHandlerContext ctx, Http2Frame frame) { if (frame instanceof Http2StreamFrame) { Http2StreamFrame streamFrame = (Http2StreamFrame) frame; ((Http2MultiplexCodecStream) streamFrame.stream()).channel.fireChildRead(streamFrame); } else if (frame instanceof Http2GoAwayFrame) { onHttp2GoAwayFrame(ctx, (Http2GoAwayFrame) frame); ctx.fireChannelRead(frame); } else if (frame instanceof Http2SettingsFrame) { Http2Settings settings = ((Http2SettingsFrame) frame).settings(); if (settings.initialWindowSize() != null) { initialOutboundStreamWindow = settings.initialWindowSize(); } ctx.fireChannelRead(frame); } else { ctx.fireChannelRead(frame); } } Http2MultiplexCodec(Http2ConnectionEncoder encoder,
                        Http2ConnectionDecoder decoder,
                        Http2Settings initialSettings,
                        ChannelHandler inboundStreamHandler,
                        ChannelHandler upgradeStreamHandler); }
Http2MultiplexCodec extends Http2FrameCodec { @Override final void onHttp2Frame(ChannelHandlerContext ctx, Http2Frame frame) { if (frame instanceof Http2StreamFrame) { Http2StreamFrame streamFrame = (Http2StreamFrame) frame; ((Http2MultiplexCodecStream) streamFrame.stream()).channel.fireChildRead(streamFrame); } else if (frame instanceof Http2GoAwayFrame) { onHttp2GoAwayFrame(ctx, (Http2GoAwayFrame) frame); ctx.fireChannelRead(frame); } else if (frame instanceof Http2SettingsFrame) { Http2Settings settings = ((Http2SettingsFrame) frame).settings(); if (settings.initialWindowSize() != null) { initialOutboundStreamWindow = settings.initialWindowSize(); } ctx.fireChannelRead(frame); } else { ctx.fireChannelRead(frame); } } Http2MultiplexCodec(Http2ConnectionEncoder encoder,
                        Http2ConnectionDecoder decoder,
                        Http2Settings initialSettings,
                        ChannelHandler inboundStreamHandler,
                        ChannelHandler upgradeStreamHandler); @Override void onHttpClientUpgrade(); @Override final void handlerAdded0(ChannelHandlerContext ctx); @Override final void handlerRemoved0(ChannelHandlerContext ctx); @Override final void channelReadComplete(ChannelHandlerContext ctx); @Override final void channelRead(ChannelHandlerContext ctx, Object msg); }
Http2MultiplexCodec extends Http2FrameCodec { @Override final void onHttp2Frame(ChannelHandlerContext ctx, Http2Frame frame) { if (frame instanceof Http2StreamFrame) { Http2StreamFrame streamFrame = (Http2StreamFrame) frame; ((Http2MultiplexCodecStream) streamFrame.stream()).channel.fireChildRead(streamFrame); } else if (frame instanceof Http2GoAwayFrame) { onHttp2GoAwayFrame(ctx, (Http2GoAwayFrame) frame); ctx.fireChannelRead(frame); } else if (frame instanceof Http2SettingsFrame) { Http2Settings settings = ((Http2SettingsFrame) frame).settings(); if (settings.initialWindowSize() != null) { initialOutboundStreamWindow = settings.initialWindowSize(); } ctx.fireChannelRead(frame); } else { ctx.fireChannelRead(frame); } } Http2MultiplexCodec(Http2ConnectionEncoder encoder,
                        Http2ConnectionDecoder decoder,
                        Http2Settings initialSettings,
                        ChannelHandler inboundStreamHandler,
                        ChannelHandler upgradeStreamHandler); @Override void onHttpClientUpgrade(); @Override final void handlerAdded0(ChannelHandlerContext ctx); @Override final void handlerRemoved0(ChannelHandlerContext ctx); @Override final void channelReadComplete(ChannelHandlerContext ctx); @Override final void channelRead(ChannelHandlerContext ctx, Object msg);  }