upgradesPipelineInSameMethodInvocation() — netty Function Reference

Source Code

codec-http/src/test/java/io/netty/handler/codec/http/HttpServerUpgradeHandlerTest.java lines 70–144

    @Test
    public void upgradesPipelineInSameMethodInvocation() {
        final HttpServerCodec httpServerCodec = new HttpServerCodec();
        final UpgradeCodecFactory factory = new UpgradeCodecFactory() {
            @Override
            public UpgradeCodec newUpgradeCodec(CharSequence protocol) {
                return new TestUpgradeCodec();
            }
        };

        ChannelHandler testInStackFrame = new ChannelDuplexHandler() {
            // marker boolean to signal that we're in the `channelRead` method
            private boolean inReadCall;
            private boolean writeUpgradeMessage;
            private boolean writeFlushed;

            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                assertFalse(inReadCall);
                assertFalse(writeUpgradeMessage);

                inReadCall = true;
                try {
                    super.channelRead(ctx, msg);
                    // All in the same call stack, the upgrade codec should receive the message,
                    // written the upgrade response, and upgraded the pipeline.
                    assertTrue(writeUpgradeMessage);
                    assertFalse(writeFlushed);
                    assertNull(ctx.pipeline().get(HttpServerCodec.class));
                    assertNotNull(ctx.pipeline().get("marker"));
                } finally {
                    inReadCall = false;
                }
            }

            @Override
            public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
                // We ensure that we're in the read call and defer the write so we can
                // make sure the pipeline was reformed irrespective of the flush completing.
                assertTrue(inReadCall);
                writeUpgradeMessage = true;
                ctx.channel().eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        ctx.write(msg, promise);
                    }
                });
                promise.addListener(future -> writeFlushed = true);
            }
        };

        HttpServerUpgradeHandler upgradeHandler = new HttpServerUpgradeHandler(httpServerCodec, factory);

        EmbeddedChannel channel = new EmbeddedChannel(testInStackFrame, httpServerCodec, upgradeHandler);

        String upgradeString = "GET / HTTP/1.1\r\n" +
            "Host: example.com\r\n" +
            "Connection: Upgrade, HTTP2-Settings\r\n" +
            "Upgrade: nextprotocol\r\n" +
            "HTTP2-Settings: AAMAAABkAAQAAP__\r\n\r\n";
        ByteBuf upgrade = Unpooled.copiedBuffer(upgradeString, CharsetUtil.US_ASCII);

        assertFalse(channel.writeInbound(upgrade));
        assertNull(channel.pipeline().get(HttpServerCodec.class));
        assertNotNull(channel.pipeline().get("marker"));

        channel.flushOutbound();
        ByteBuf upgradeMessage = channel.readOutbound();
        String expectedHttpResponse = "HTTP/1.1 101 Switching Protocols\r\n" +
            "connection: upgrade\r\n" +
            "upgrade: nextprotocol\r\n\r\n";
        assertEquals(expectedHttpResponse, upgradeMessage.toString(CharsetUtil.US_ASCII));
        assertTrue(upgradeMessage.release());
        assertFalse(channel.finishAndReleaseAll());
    }