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

@Test
public void testFullHttpRequest() {
    // test that ContentDecoder can be used after the ObjectAggregator
    HttpRequestDecoder decoder = new HttpRequestDecoder(4096, 4096, 5);
    HttpObjectAggregator aggregator = new HttpObjectAggregator(1024);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, aggregator, decompressor);
    String headers = "POST / HTTP/1.1\r\n" +
                     "Content-Length: " + GZ_HELLO_WORLD.length + "\r\n" +
                     "Content-Encoding: gzip\r\n" +
                     "\r\n";
    assertTrue(channel.writeInbound(Unpooled.copiedBuffer(headers.getBytes(), GZ_HELLO_WORLD)));

    Queue<Object> req = channel.inboundMessages();
    assertTrue(req.size() > 1);
    int contentLength = 0;
    contentLength = calculateContentLength(req, contentLength);

    byte[] receivedContent = readContent(req, contentLength, true);

    assertEquals(HELLO_WORLD, new String(receivedContent, CharsetUtil.US_ASCII));

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}
 

@Test
public void testFullHttpResponse() {
    // test that ContentDecoder can be used after the ObjectAggregator
    HttpResponseDecoder decoder = new HttpResponseDecoder(4096, 4096, 5);
    HttpObjectAggregator aggregator = new HttpObjectAggregator(1024);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, aggregator, decompressor);
    String headers = "HTTP/1.1 200 OK\r\n" +
                     "Content-Length: " + GZ_HELLO_WORLD.length + "\r\n" +
                     "Content-Encoding: gzip\r\n" +
                     "\r\n";
    assertTrue(channel.writeInbound(Unpooled.copiedBuffer(headers.getBytes(), GZ_HELLO_WORLD)));

    Queue<Object> resp = channel.inboundMessages();
    assertTrue(resp.size() > 1);
    int contentLength = 0;
    contentLength = calculateContentLength(resp, contentLength);

    byte[] receivedContent = readContent(resp, contentLength, true);

    assertEquals(HELLO_WORLD, new String(receivedContent, CharsetUtil.US_ASCII));

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}
 

@Test
public void testFullHttpResponseEOF() {
    // test that ContentDecoder can be used after the ObjectAggregator
    HttpResponseDecoder decoder = new HttpResponseDecoder(4096, 4096, 5);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor);
    String headers = "HTTP/1.1 200 OK\r\n" +
            "Content-Encoding: gzip\r\n" +
            "\r\n";
    assertTrue(channel.writeInbound(Unpooled.copiedBuffer(headers.getBytes(), GZ_HELLO_WORLD)));
    // This should terminate it.
    assertTrue(channel.finish());

    Queue<Object> resp = channel.inboundMessages();
    assertTrue(resp.size() > 1);
    int contentLength = 0;
    contentLength = calculateContentLength(resp, contentLength);

    byte[] receivedContent = readContent(resp, contentLength, false);

    assertEquals(HELLO_WORLD, new String(receivedContent, CharsetUtil.US_ASCII));

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}
 

@Test
// verifies backward compatibility
public void testConcatenatedStreamsReadFirstOnly() throws IOException {
    EmbeddedChannel chDecoderGZip = new EmbeddedChannel(createDecoder(ZlibWrapper.GZIP));

    try {
        byte[] bytes = IOUtils.toByteArray(getClass().getResourceAsStream("/multiple.gz"));

        assertTrue(chDecoderGZip.writeInbound(Unpooled.copiedBuffer(bytes)));
        Queue<Object> messages = chDecoderGZip.inboundMessages();
        assertEquals(1, messages.size());

        ByteBuf msg = (ByteBuf) messages.poll();
        assertEquals("a", msg.toString(CharsetUtil.UTF_8));
        ReferenceCountUtil.release(msg);
    } finally {
        assertFalse(chDecoderGZip.finish());
        chDecoderGZip.close();
    }
}
 

@Test
public void testConcatenatedStreamsReadFully() throws IOException {
    EmbeddedChannel chDecoderGZip = new EmbeddedChannel(new JdkZlibDecoder(true));

    try {
        byte[] bytes = IOUtils.toByteArray(getClass().getResourceAsStream("/multiple.gz"));

        assertTrue(chDecoderGZip.writeInbound(Unpooled.copiedBuffer(bytes)));
        Queue<Object> messages = chDecoderGZip.inboundMessages();
        assertEquals(2, messages.size());

        for (String s : Arrays.asList("a", "b")) {
            ByteBuf msg = (ByteBuf) messages.poll();
            assertEquals(s, msg.toString(CharsetUtil.UTF_8));
            ReferenceCountUtil.release(msg);
        }
    } finally {
        assertFalse(chDecoderGZip.finish());
        chDecoderGZip.close();
    }
}
 

@Test
public void testRequestContentLength1() {
    // case 1: test that ContentDecompressor either sets the correct Content-Length header
    // or removes it completely (handlers down the chain must rely on LastHttpContent object)

    // force content to be in more than one chunk (5 bytes/chunk)
    HttpRequestDecoder decoder = new HttpRequestDecoder(4096, 4096, 5);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor);
    String headers = "POST / HTTP/1.1\r\n" +
                     "Content-Length: " + GZ_HELLO_WORLD.length + "\r\n" +
                     "Content-Encoding: gzip\r\n" +
                     "\r\n";
    ByteBuf buf = Unpooled.copiedBuffer(headers.getBytes(CharsetUtil.US_ASCII), GZ_HELLO_WORLD);
    assertTrue(channel.writeInbound(buf));

    Queue<Object> req = channel.inboundMessages();
    assertTrue(req.size() >= 1);
    Object o = req.peek();
    assertThat(o, is(instanceOf(HttpRequest.class)));
    HttpRequest r = (HttpRequest) o;
    String v = r.headers().get(HttpHeaderNames.CONTENT_LENGTH);
    Long value = v == null ? null : Long.parseLong(v);
    assertTrue(value == null || value.longValue() == HELLO_WORLD.length());

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}
 

@Test
public void testResponseContentLength1() {
    // case 1: test that ContentDecompressor either sets the correct Content-Length header
    // or removes it completely (handlers down the chain must rely on LastHttpContent object)

    // force content to be in more than one chunk (5 bytes/chunk)
    HttpResponseDecoder decoder = new HttpResponseDecoder(4096, 4096, 5);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor);
    String headers = "HTTP/1.1 200 OK\r\n" +
                     "Content-Length: " + GZ_HELLO_WORLD.length + "\r\n" +
                     "Content-Encoding: gzip\r\n" +
                     "\r\n";
    ByteBuf buf = Unpooled.copiedBuffer(headers.getBytes(CharsetUtil.US_ASCII), GZ_HELLO_WORLD);
    assertTrue(channel.writeInbound(buf));

    Queue<Object> resp = channel.inboundMessages();
    assertTrue(resp.size() >= 1);
    Object o = resp.peek();
    assertThat(o, is(instanceOf(HttpResponse.class)));
    HttpResponse r = (HttpResponse) o;

    assertFalse("Content-Length header not removed.", r.headers().contains(HttpHeaderNames.CONTENT_LENGTH));

    String transferEncoding = r.headers().get(HttpHeaderNames.TRANSFER_ENCODING);
    assertNotNull("Content-length as well as transfer-encoding not set.", transferEncoding);
    assertEquals("Unexpected transfer-encoding value.", HttpHeaderValues.CHUNKED.toString(), transferEncoding);

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}
 

@Test
public void testRequestContentLength1() {
    // case 1: test that ContentDecompressor either sets the correct Content-Length header
    // or removes it completely (handlers down the chain must rely on LastHttpContent object)

    // force content to be in more than one chunk (5 bytes/chunk)
    HttpRequestDecoder decoder = new HttpRequestDecoder(4096, 4096, 5);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor);
    String headers = "POST / HTTP/1.1\r\n" +
                     "Content-Length: " + GZ_HELLO_WORLD.length + "\r\n" +
                     "Content-Encoding: gzip\r\n" +
                     "\r\n";
    ByteBuf buf = Unpooled.copiedBuffer(headers.getBytes(CharsetUtil.US_ASCII), GZ_HELLO_WORLD);
    assertTrue(channel.writeInbound(buf));

    Queue<Object> req = channel.inboundMessages();
    assertTrue(req.size() >= 1);
    Object o = req.peek();
    assertThat(o, is(instanceOf(HttpRequest.class)));
    HttpRequest r = (HttpRequest) o;
    String v = r.headers().get(HttpHeaders.Names.CONTENT_LENGTH);
    Long value = v == null ? null : Long.parseLong(v);
    assertTrue(value == null || value.longValue() == HELLO_WORLD.length());

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}
 

@Test
public void testResponseContentLength1() {
    // case 1: test that ContentDecompressor either sets the correct Content-Length header
    // or removes it completely (handlers down the chain must rely on LastHttpContent object)

    // force content to be in more than one chunk (5 bytes/chunk)
    HttpResponseDecoder decoder = new HttpResponseDecoder(4096, 4096, 5);
    HttpContentDecoder decompressor = new HttpContentDecompressor();
    EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor);
    String headers = "HTTP/1.1 200 OK\r\n" +
                     "Content-Length: " + GZ_HELLO_WORLD.length + "\r\n" +
                     "Content-Encoding: gzip\r\n" +
                     "\r\n";
    ByteBuf buf = Unpooled.copiedBuffer(headers.getBytes(CharsetUtil.US_ASCII), GZ_HELLO_WORLD);
    assertTrue(channel.writeInbound(buf));

    Queue<Object> resp = channel.inboundMessages();
    assertTrue(resp.size() >= 1);
    Object o = resp.peek();
    assertThat(o, is(instanceOf(HttpResponse.class)));
    HttpResponse r = (HttpResponse) o;
    String v = r.headers().get(HttpHeaders.Names.CONTENT_LENGTH);
    Long value = v == null ? null : Long.parseLong(v);
    assertTrue(value == null || value.longValue() == HELLO_WORLD.length());

    assertHasInboundMessages(channel, true);
    assertHasOutboundMessages(channel, false);
    assertFalse(channel.finish());
}