listenerIsNotifiedOfGoawayBeforeStreamsAreRemovedFromTheConnection() — netty Function Reference

Source Code

codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionRoundtripTest.java lines 915–991

    @Test
    public void listenerIsNotifiedOfGoawayBeforeStreamsAreRemovedFromTheConnection() throws Exception {
        final AtomicReference<Http2Stream.State> clientStream3State = new AtomicReference<Http2Stream.State>();
        final CountDownLatch clientGoAwayLatch = new CountDownLatch(1);
        doAnswer(new Answer<Void>() {
            @Override
            public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
                clientStream3State.set(http2Client.connection().stream(3).state());
                clientGoAwayLatch.countDown();
                return null;
            }
        }).when(clientListener).onGoAwayRead(any(ChannelHandlerContext.class), anyInt(), anyLong(), any(ByteBuf.class));

        bootstrapEnv(1, 1, 3, 1, 1);

        // We want both sides to do graceful shutdown during the test.
        setClientGracefulShutdownTime(10000);
        setServerGracefulShutdownTime(10000);

        // Create a single stream by sending a HEADERS frame to the server.
        final Http2Headers headers = dummyHeaders();
        runInChannel(clientChannel, new Http2Runnable() {
            @Override
            public void run() throws Http2Exception {
                http2Client.encoder().writeHeaders(ctx(), 1, headers, 0, (short) 16, false, 0,
                    false, newPromise());
                http2Client.encoder().writeHeaders(ctx(), 3, headers, 0, (short) 16, false, 0,
                    false, newPromise());
                http2Client.flush(ctx());
            }
        });

        assertTrue(serverSettingsAckLatch.await(DEFAULT_AWAIT_TIMEOUT_SECONDS, SECONDS));

        // Server has received the headers, so the stream is open
        assertTrue(requestLatch.await(DEFAULT_AWAIT_TIMEOUT_SECONDS, SECONDS));

        runInChannel(serverChannel, new Http2Runnable() {
            @Override
            public void run() throws Http2Exception {
                http2Server.encoder().writeGoAway(serverCtx(), 1, NO_ERROR.code(), EMPTY_BUFFER, serverNewPromise());
                http2Server.flush(serverCtx());
            }
        });

        // wait for the client to receive the GO_AWAY.
        assertTrue(clientGoAwayLatch.await(DEFAULT_AWAIT_TIMEOUT_SECONDS, SECONDS));
        verify(clientListener).onGoAwayRead(any(ChannelHandlerContext.class), eq(1), eq(NO_ERROR.code()),
            any(ByteBuf.class));
        assertEquals(Http2Stream.State.OPEN, clientStream3State.get());

        // Make sure that stream 3 has been closed which is true if it's gone.
        final CountDownLatch probeStreamCount = new CountDownLatch(1);
        final AtomicBoolean stream3Exists = new AtomicBoolean();
        final AtomicInteger streamCount = new AtomicInteger();
        runInChannel(this.clientChannel, new Http2Runnable() {
            @Override
            public void run() throws Http2Exception {
                stream3Exists.set(http2Client.connection().stream(3) != null);
                streamCount.set(http2Client.connection().numActiveStreams());
                probeStreamCount.countDown();
            }
        });
        // The stream should be closed right after
        assertTrue(probeStreamCount.await(DEFAULT_AWAIT_TIMEOUT_SECONDS, SECONDS));
        assertEquals(1, streamCount.get());
        assertFalse(stream3Exists.get());

        // Wait for the server to receive a GO_AWAY, but this is expected to timeout!
        assertFalse(goAwayLatch.await(1, SECONDS));
        verify(serverListener, never()).onGoAwayRead(any(ChannelHandlerContext.class), anyInt(), anyLong(),
            any(ByteBuf.class));

        // Shutdown shouldn't wait for the server to close streams
        setClientGracefulShutdownTime(0);
        setServerGracefulShutdownTime(0);
    }