io.netty.channel.embedded.EmbeddedChannel#flush ( )源码实例Demo

private static void testUpgrade(Http2ConnectionHandler handler) {
    FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.OPTIONS, "*");
    request.headers().set(HttpHeaderNames.HOST, "netty.io");
    request.headers().set(HttpHeaderNames.CONNECTION, "Upgrade, HTTP2-Settings");
    request.headers().set(HttpHeaderNames.UPGRADE, "h2c");
    request.headers().set("HTTP2-Settings", "AAMAAABkAAQAAP__");

    EmbeddedChannel channel = new EmbeddedChannel(new ChannelInboundHandlerAdapter());
    ChannelHandlerContext ctx = channel.pipeline().firstContext();
    Http2ServerUpgradeCodec codec = new Http2ServerUpgradeCodec("connectionHandler", handler);
    assertTrue(codec.prepareUpgradeResponse(ctx, request, new DefaultHttpHeaders()));
    codec.upgradeTo(ctx, request);
    // Flush the channel to ensure we write out all buffered data
    channel.flush();

    assertSame(handler, channel.pipeline().remove("connectionHandler"));
    assertNull(channel.pipeline().get(handler.getClass()));
    assertTrue(channel.finish());

    // Check that the preface was send (a.k.a the settings frame)
    ByteBuf settingsBuffer = channel.readOutbound();
    assertNotNull(settingsBuffer);
    settingsBuffer.release();

    assertNull(channel.readOutbound());
}
 

@Test
public void testFlushViaReadComplete() {
    final AtomicInteger flushCount = new AtomicInteger();
    EmbeddedChannel channel = newChannel(flushCount, false);
    // Flush should go through as there is no read loop in progress.
    channel.flush();
    channel.runPendingTasks();
    assertEquals(1, flushCount.get());

    // Simulate read loop;
    channel.pipeline().fireChannelRead(1L);
    assertEquals(1, flushCount.get());
    channel.pipeline().fireChannelRead(2L);
    assertEquals(1, flushCount.get());
    assertNull(channel.readOutbound());
    channel.pipeline().fireChannelReadComplete();
    assertEquals(2, flushCount.get());
    // Now flush again as the read loop is complete.
    channel.flush();
    channel.runPendingTasks();
    assertEquals(3, flushCount.get());
    assertEquals(1L, channel.readOutbound());
    assertEquals(2L, channel.readOutbound());
    assertNull(channel.readOutbound());
    assertFalse(channel.finish());
}
 

@Test
public void testFlush() {
    Lz4FrameEncoder encoder = new Lz4FrameEncoder();
    EmbeddedChannel channel = new EmbeddedChannel(encoder);
    int size = 27;
    ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(size, size);
    buf.writerIndex(size);
    Assert.assertEquals(0, encoder.getBackingBuffer().readableBytes());
    channel.write(buf);
    Assert.assertTrue(channel.outboundMessages().isEmpty());
    Assert.assertEquals(size, encoder.getBackingBuffer().readableBytes());
    channel.flush();
    Assert.assertTrue(channel.finish());
    Assert.assertTrue(channel.releaseOutbound());
    Assert.assertFalse(channel.releaseInbound());
}
 

@Test
public void testAllocatingAroundBlockSize() {
    int blockSize = 100;
    Lz4FrameEncoder encoder = newEncoder(blockSize, Lz4FrameEncoder.DEFAULT_MAX_ENCODE_SIZE);
    EmbeddedChannel channel = new EmbeddedChannel(encoder);

    int size = blockSize - 1;
    ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(size, size);
    buf.writerIndex(size);
    Assert.assertEquals(0, encoder.getBackingBuffer().readableBytes());
    channel.write(buf);
    Assert.assertEquals(size, encoder.getBackingBuffer().readableBytes());

    int nextSize = size - 1;
    buf = ByteBufAllocator.DEFAULT.buffer(nextSize, nextSize);
    buf.writerIndex(nextSize);
    channel.write(buf);
    Assert.assertEquals(size + nextSize - blockSize, encoder.getBackingBuffer().readableBytes());

    channel.flush();
    Assert.assertEquals(0, encoder.getBackingBuffer().readableBytes());
    Assert.assertTrue(channel.finish());
    Assert.assertTrue(channel.releaseOutbound());
    Assert.assertFalse(channel.releaseInbound());
}
 

@Test
public void testNullValuesAddToLength() throws Exception {
    EmbeddedChannel channel = new EmbeddedChannel();
    Messages.sendDataRow(
        channel,
        new RowN($(10, null)),
        Arrays.asList(PGTypes.get(DataTypes.INTEGER), PGTypes.get(DataTypes.STRING)), null
    );
    channel.flush();
    ByteBuf buffer = channel.readOutbound();

    try {
        // message type
        assertThat((char) buffer.readByte(), is('D'));

        // size of the message
        assertThat(buffer.readInt(), is(16));
        assertThat(buffer.readableBytes(), is(12)); // 16 - INT4 because the size was already read
    } finally {
        buffer.release();
        channel.finishAndReleaseAll();
    }
}
 

private static void testUpgrade(Http2ConnectionHandler handler) throws Exception {
    FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.OPTIONS, "*");

    EmbeddedChannel channel = new EmbeddedChannel(new ChannelInboundHandlerAdapter());
    ChannelHandlerContext ctx = channel.pipeline().firstContext();
    Http2ClientUpgradeCodec codec = new Http2ClientUpgradeCodec("connectionHandler", handler);
    codec.setUpgradeHeaders(ctx, request);
    // Flush the channel to ensure we write out all buffered data
    channel.flush();

    codec.upgradeTo(ctx, null);
    assertNotNull(channel.pipeline().get("connectionHandler"));

    assertTrue(channel.finishAndReleaseAll());
}
 

@Test
public void testFlushViaScheduledTask() {
    final AtomicInteger flushCount = new AtomicInteger();
    EmbeddedChannel channel = newChannel(flushCount,  true);
    // Flushes should not go through immediately, as they're scheduled as an async task
    channel.flush();
    assertEquals(0, flushCount.get());
    channel.flush();
    assertEquals(0, flushCount.get());
    // Trigger the execution of the async task
    channel.runPendingTasks();
    assertEquals(1, flushCount.get());
    assertFalse(channel.finish());
}
 

@Test
public void testFlushViaThresholdOutsideOfReadLoop() {
    final AtomicInteger flushCount = new AtomicInteger();
    EmbeddedChannel channel = newChannel(flushCount, true);
    // After a given threshold, the async task should be bypassed and a flush should be triggered immediately
    for (int i = 0; i < EXPLICIT_FLUSH_AFTER_FLUSHES; i++) {
        channel.flush();
    }
    assertEquals(1, flushCount.get());
    assertFalse(channel.finish());
}
 

@Test
public void testImmediateFlushOutsideOfReadLoop() {
    final AtomicInteger flushCount = new AtomicInteger();
    EmbeddedChannel channel = newChannel(flushCount, false);
    channel.flush();
    assertEquals(1, flushCount.get());
    assertFalse(channel.finish());
}
 

@Test
public void testRemoveAndWriteAllReentrance() {
    EmbeddedChannel channel = newChannel();
    final PendingWriteQueue queue = new PendingWriteQueue(channel.pipeline().firstContext());

    ChannelPromise promise = channel.newPromise();
    promise.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            queue.removeAndWriteAll();
        }
    });
    queue.add(1L, promise);

    ChannelPromise promise2 = channel.newPromise();
    queue.add(2L, promise2);
    queue.removeAndWriteAll();
    channel.flush();
    assertTrue(promise.isSuccess());
    assertTrue(promise2.isSuccess());
    assertTrue(channel.finish());

    assertEquals(1L, channel.readOutbound());
    assertEquals(2L, channel.readOutbound());
    assertNull(channel.readOutbound());
    assertNull(channel.readInbound());
}
 

@Test
public void testMixedWritability() {
    final StringBuilder buf = new StringBuilder();
    EmbeddedChannel ch = new EmbeddedChannel(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
            buf.append(ctx.channel().isWritable());
            buf.append(' ');
        }
    });

    ch.config().setWriteBufferLowWaterMark(128);
    ch.config().setWriteBufferHighWaterMark(256);

    ChannelOutboundBuffer cob = ch.unsafe().outboundBuffer();

    // Trigger channelWritabilityChanged() by writing a lot.
    ch.write(buffer().writeZero(257));
    assertThat(buf.toString(), is("false "));

    // Ensure that setting a user-defined writability flag to false does not trigger channelWritabilityChanged()
    cob.setUserDefinedWritability(1, false);
    ch.runPendingTasks();
    assertThat(buf.toString(), is("false "));

    // Ensure reducing the totalPendingWriteBytes down to zero does not trigger channelWritabilityChanged()
    // because of the user-defined writability flag.
    ch.flush();
    assertThat(cob.totalPendingWriteBytes(), is(0L));
    assertThat(buf.toString(), is("false "));

    // Ensure that setting the user-defined writability flag to true triggers channelWritabilityChanged()
    cob.setUserDefinedWritability(1, true);
    ch.runPendingTasks();
    assertThat(buf.toString(), is("false true "));

    safeClose(ch);
}
 

@Test
public void testRemoveAndWriteAllReentrance() {
    EmbeddedChannel channel = new EmbeddedChannel(new ChannelInboundHandlerAdapter());
    final PendingWriteQueue queue = new PendingWriteQueue(channel.pipeline().firstContext());

    ChannelPromise promise = channel.newPromise();
    promise.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            queue.removeAndWriteAll();
        }
    });
    queue.add(1L, promise);

    ChannelPromise promise2 = channel.newPromise();
    queue.add(2L, promise2);
    queue.removeAndWriteAll();
    channel.flush();
    assertTrue(promise.isSuccess());
    assertTrue(promise2.isSuccess());
    assertTrue(channel.finish());

    assertEquals(1L, channel.readOutbound());
    assertEquals(2L, channel.readOutbound());
    assertNull(channel.readOutbound());
    assertNull(channel.readInbound());
}
 

@Test
public void testMixedWritability() {
    final StringBuilder buf = new StringBuilder();
    EmbeddedChannel ch = new EmbeddedChannel(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
            buf.append(ctx.channel().isWritable());
            buf.append(' ');
        }
    });

    ch.config().setWriteBufferLowWaterMark(128);
    ch.config().setWriteBufferHighWaterMark(256);

    ChannelOutboundBuffer cob = ch.unsafe().outboundBuffer();

    // Trigger channelWritabilityChanged() by writing a lot.
    ch.write(buffer().writeZero(256));
    assertThat(buf.toString(), is("false "));

    // Ensure that setting a user-defined writability flag to false does not trigger channelWritabilityChanged()
    cob.setUserDefinedWritability(1, false);
    ch.runPendingTasks();
    assertThat(buf.toString(), is("false "));

    // Ensure reducing the totalPendingWriteBytes down to zero does not trigger channelWritabilityChannged()
    // because of the user-defined writability flag.
    ch.flush();
    assertThat(cob.totalPendingWriteBytes(), is(0L));
    assertThat(buf.toString(), is("false "));

    // Ensure that setting the user-defined writability flag to true triggers channelWritabilityChanged()
    cob.setUserDefinedWritability(1, true);
    ch.runPendingTasks();
    assertThat(buf.toString(), is("false true "));

    safeClose(ch);
}
 

private void testCompress0(ZlibWrapper encoderWrapper, ZlibWrapper decoderWrapper, ByteBuf data) throws Exception {
    EmbeddedChannel chEncoder = new EmbeddedChannel(createEncoder(encoderWrapper));
    EmbeddedChannel chDecoderZlib = new EmbeddedChannel(createDecoder(decoderWrapper));

    try {
        chEncoder.writeOutbound(data.retain());
        chEncoder.flush();
        data.resetReaderIndex();

        for (;;) {
            ByteBuf deflatedData = chEncoder.readOutbound();
            if (deflatedData == null) {
                break;
            }
            chDecoderZlib.writeInbound(deflatedData);
        }

        byte[] decompressed = new byte[data.readableBytes()];
        int offset = 0;
        for (;;) {
            ByteBuf buf = chDecoderZlib.readInbound();
            if (buf == null) {
                break;
            }
            int length = buf.readableBytes();
            buf.readBytes(decompressed, offset, length);
            offset += length;
            buf.release();
            if (offset == decompressed.length) {
                break;
            }
        }
        assertEquals(data, Unpooled.wrappedBuffer(decompressed));
        assertNull(chDecoderZlib.readInbound());

        // Closing an encoder channel will generate a footer.
        assertTrue(chEncoder.finish());
        for (;;) {
            Object msg = chEncoder.readOutbound();
            if (msg == null) {
                break;
            }
            ReferenceCountUtil.release(msg);
        }
        // But, the footer will be decoded into nothing. It's only for validation.
        assertFalse(chDecoderZlib.finish());

        data.release();
    } finally {
        dispose(chEncoder);
        dispose(chDecoderZlib);
    }
}
 

private void testCompress0(ZlibWrapper encoderWrapper, ZlibWrapper decoderWrapper, ByteBuf data) throws Exception {
    EmbeddedChannel chEncoder = new EmbeddedChannel(createEncoder(encoderWrapper));
    EmbeddedChannel chDecoderZlib = new EmbeddedChannel(createDecoder(decoderWrapper));

    try {
        chEncoder.writeOutbound(data.copy());
        chEncoder.flush();

        for (;;) {
            ByteBuf deflatedData = (ByteBuf) chEncoder.readOutbound();
            if (deflatedData == null) {
                break;
            }
            chDecoderZlib.writeInbound(deflatedData);
        }

        byte[] decompressed = new byte[data.readableBytes()];
        int offset = 0;
        for (;;) {
            ByteBuf buf = (ByteBuf) chDecoderZlib.readInbound();
            if (buf == null) {
                break;
            }
            int length = buf.readableBytes();
            buf.readBytes(decompressed, offset, length);
            offset += length;
            buf.release();
            if (offset == decompressed.length) {
                break;
            }
        }
        assertEquals(data, Unpooled.wrappedBuffer(decompressed));
        assertNull(chDecoderZlib.readInbound());

        // Closing an encoder channel will generate a footer.
        assertTrue(chEncoder.finish());
        for (;;) {
            Object msg = chEncoder.readOutbound();
            if (msg == null) {
                break;
            }
            ReferenceCountUtil.release(msg);
        }
        // But, the footer will be decoded into nothing. It's only for validation.
        assertFalse(chDecoderZlib.finish());

        data.release();
    } finally {
        dispose(chEncoder);
        dispose(chDecoderZlib);
    }
}