Repository: netty/netty
Issue: HttpToHttp2ConnectionHandler does not support converting from Http(Message|Content) to Http2 frames

Is there a reason why `HttpToHttp2ConnectionHandler` does not support converting from `Http(Message|Content)` to Http2 frames?  Is there something that the handler requires that it cannot retrieve?  Or some data missing from `Http(Message|Content)` that is required?

My use case is that I have an `HttpContentCompressor` further down the pipeline, which ends up creating `HttpMessage` and `HttpContent` objects and sending them to `HttpToHttp2ConnectionHandler` which is then not being converted to Http2 frames.

For now I have extended `HttpToHttp2ConnectionHandler` and overridden the `write(...)` to support `Http(Message|Content)`, but I think something like this should be in Netty directly.


Issue comments:
/cc @Scottmitch 

Looks like `HttpToHttp2ConnectionHandler` currently requires `FullHttpMessage`.  I don't recall whether or not there was a reason that it couldn't take `HttpMessage` or `HttpContent`.  Contributions are always welcome :) 

I think the main driver was HTTP generally has no proper framing and no stream concept. Without building higher level abstractions it is difficult to correlate one `HttpContent` with a particular stream.  `FullHttpMessage` provides the accumulation of all data and definitively associates all HTTP objects with a stream ID.

We may be able to keep state in `HttpToHttp2ConnectionHandler` as to the current stream ID or create some kind of interface which combines `HttpMessage` and something that provides a stream id (i.e `int http2StreamId()`), but I haven't thought about it too much.

I can think more when I get some more cycles...but as @nmittler said contributions are welcome :)

I'm more than willing to help out with this.  As I've stated previously, I have already implemented a rough draft that caters for my scenario.

@Scottmitch addressing the lack of a streaming concept, could we take the same approach used in the SPDY handlers by tracking the current stream ID? (see [here](https://github.com/netty/netty/blob/master/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java#L121) [here](https://github.com/netty/netty/blob/master/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java#L163) and [here](https://github.com/netty/netty/blob/master/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java#L278))

Other question I have is what do we do about trailing headers?  It doesn't seem like the current HTTP/2 implementation converts `FullHttpMessage` HTTP/1 trailing headers into some HTTP/2 Frame (I assume we'd use a HEADERS frame as specified in [HTTP Request/Response Exchange](https://http2.github.io/http2-spec/#rfc.section.8.1) point 4.).

> @Scottmitch addressing the lack of a streaming concept, could we take the same approach used in the SPDY handlers by tracking the current stream ID? (see here here and here)

Yes I think that is inline with what I was thinking.

> Other question I have is what do we do about trailing headers? It doesn't seem like the current HTTP/2 implementation converts FullHttpMessage HTTP/1 trailing headers into some HTTP/2 Frame (I assume we'd use a HEADERS frame as specified in HTTP Request/Response Exchange point 4.).

Yes if we are ignoring trailing headers then that is a bug.  The HTTP/2 codec should support this just by doing a `writeHeaders(..)` call. We should just take care to end the stream at the appropriate time.

Sounds good. Any takers?

> Am 09.07.2015 um 03:14 schrieb Scott Mitchell notifications@github.com:
> 
> @Scottmitch addressing the lack of a streaming concept, could we take the same approach used in the SPDY handlers by tracking the current stream ID? (see here here and here)
> 
> Yes I think that is inline with what I was thinking.
> 
> Other question I have is what do we do about trailing headers? It doesn't seem like the current HTTP/2 implementation converts FullHttpMessage HTTP/1 trailing headers into some HTTP/2 Frame (I assume we'd use a HEADERS frame as specified in HTTP Request/Response Exchange point 4.).
> 
> Yes if we are ignoring trailing headers then that is a bug. The HTTP/2 codec should support this just by doing a writeHeaders(..) call. We should just take care to end the stream at the appropriate time.
> 
> —
> Reply to this email directly or view it on GitHub.

I can give this a shot tonight.  I'll let you know how I get on.

Hey guys,

Sorry its taken so long... [here it is](https://github.com/netty/netty/pull/3975)

Closing this issue now that the pull request has been cherry-picked into master (3ec02bf) and 4.1 (57d28dd)


Pull request review comments:
Should probably set `release = false` just before the write.

Just to aid in readability, can you separate these blocks out into separate methods (e.g. `writeHttpMessage` and `writeHttpContent`)?

I think we should be able to simplify the logic a bit.  Could we do something like this:

``` java
HttpContent content = null;
// Just always use the promise aggregator.
SimpleChannelPromiseAggregator promiseAggregator = new SimpleChannelPromiseAggregator(promise, ctx.channel, ctx.executor);

if (msg instanceof HttpMessage) {
    HttpMessage httpMsg = (HttpMessage) msg;

    // Write headers using the promise aggregator.
}

if (msg instanceof HttpContent) {
    HttpContent httpContent = (HttpContent) msg;

    // Write the data.

    if (msg instanceof LastHttpContent) {
        LastHttpContent lastHttpContent = (LastHttpContent) msg;

        // Write Data

        // Write trailers if provided.
    }
}

promiseAggregator.doneAllocatingPromises();
```

WDYT?

+1

I think after is good so as writeData() may throw an exception before write any data.

I think the cast is not needed

yeah maybe not a bad idea

toLowerCase(Locale.US) ?

Change signature to accept HttpHeaders ?

+1. Less duplication and we will only have to worry about releasing in the `HttpContent` block where you would expect it.

I'm not sure we should check this here (or worry about resetting `currentStreamId = 0`).  The HTTP/2 codec is responsible for enforcing this.

+1 to @nmittler. the writeData is responsible for releasing the data (even in the presence of exceptions) per the interface contract.  We should avoid double releasing here and elsewhere in this PR.

I think @nmittler's suggestion (https://github.com/netty/netty/pull/3975#discussion_r34686595) to consolidate the code will help ensure we are releasing correctly.

Lets remove this per https://github.com/netty/netty/pull/3975#discussion-diff-34731790

+1. This code should also be shared with the other method `toHttp2Headers` that also has this code.

I'm not sure this is necessary. It is converted to an `AsciiString` which implies a certain encoding already. WDYT?

Need to call `promiseAggregator.donAllocatingPromises()` here too ... above `setFailure`.

Maybe further simplify the logic here to something like:

``` java
if (msg instanceof HttpMessage) {
    // Write headers.
}

if (!endStream && msg instanceof HttpContent) {
    // If this is the last content, extract the trailers.
    boolean lastContent = false;
    Http2Headers trailers = null;
    if (msg instanceof LastHttpContent) {
        lastContent = true;
        trailers = HttpUtil.toHttp2Headers(lastContent.trailingHeaders());
        // TODO: check to see if trailers can really be null here.
        if (trailers == null || trailers.isEmpty()) {
            trailers = null;
        }
    }

    release = false;
    encoder.writeData(..., lastContent && trailers == null, ...);
    if (trailers != null) {
        encoder.writeHeaders(..., true, ...);
    }

    endStream = lastContent;
}
```

@nmittler - I don't think we should have to call `donAllocatingPromises()` in the event we are able to detect an immediate failure. I think the `SimpleChannelPromiseAggregator` should support termination by calling `setFailure(..)`.

+1. I think we can drop the `null` check for `trailers` too.

Ah ok ... wasn't entirely sure.  Haven't looked at that class in a while :)

hmm... should this always end the stream if a FullHttpMessage is used ?

I would move the release directly before writeData(...) call. Just in case

As far as I'm aware, yes.  If the FullHttpMessage don't not contain any data, the only HTTP/2 frame to write is the HEADERS frame.  You can see the old code did the same thing.

You are right.  I'll fix that.

nit: just return here.

``` java
return out.add(toHttp2Headers(inHeaders));
```


Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandler.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandler.java
index ca3676617d8..6bb45045dcf 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandler.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandler.java
@@ -15,11 +15,16 @@
 
 package io.netty.handler.codec.http2;
 
+import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.FullHttpMessage;
+import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMessage;
+import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.codec.http2.Http2CodecUtil.SimpleChannelPromiseAggregator;
+import io.netty.util.ReferenceCountUtil;
 
 /**
  * Translates HTTP/1.x object writes into HTTP/2 frames.
@@ -27,6 +32,9 @@
  * See {@link InboundHttp2ToHttpAdapter} to get translation from HTTP/2 frames to HTTP/1.x objects.
  */
 public class HttpToHttp2ConnectionHandler extends Http2ConnectionHandler {
+
+    private int currentStreamId;
+
     public HttpToHttp2ConnectionHandler(boolean server, Http2FrameListener listener) {
         super(server, listener);
     }
@@ -57,45 +65,64 @@ private int getStreamId(HttpHeaders httpHeaders) throws Exception {
     }
 
     /**
-     * Handles conversion of a {@link FullHttpMessage} to HTTP/2 frames.
+     * Handles conversion of {@link HttpMessage} and {@link HttpContent} to HTTP/2 frames.
      */
     @Override
     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
-        if (msg instanceof FullHttpMessage) {
-            FullHttpMessage httpMsg = (FullHttpMessage) msg;
-            boolean hasData = httpMsg.content().isReadable();
-            boolean httpMsgNeedRelease = true;
-            SimpleChannelPromiseAggregator promiseAggregator = null;
-            try {
+
+        if (!(msg instanceof HttpMessage || msg instanceof HttpContent)) {
+            ctx.write(msg, promise);
+            return;
+        }
+
+        boolean release = true;
+        SimpleChannelPromiseAggregator promiseAggregator =
+                new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
+        try {
+            Http2ConnectionEncoder encoder = encoder();
+            boolean endStream = false;
+            if (msg instanceof HttpMessage) {
+                final HttpMessage httpMsg = (HttpMessage) msg;
+
                 // Provide the user the opportunity to specify the streamId
-                int streamId = getStreamId(httpMsg.headers());
+                currentStreamId = getStreamId(httpMsg.headers());
 
                 // Convert and write the headers.
                 Http2Headers http2Headers = HttpUtil.toHttp2Headers(httpMsg);
-                Http2ConnectionEncoder encoder = encoder();
-
-                if (hasData) {
-                    promiseAggregator = new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
-                    encoder.writeHeaders(ctx, streamId, http2Headers, 0, false, promiseAggregator.newPromise());
-                    httpMsgNeedRelease = false;
-                    encoder.writeData(ctx, streamId, httpMsg.content(), 0, true, promiseAggregator.newPromise());
-                    promiseAggregator.doneAllocatingPromises();
-                } else {
-                    encoder.writeHeaders(ctx, streamId, http2Headers, 0, true, promise);
-                }
-            } catch (Throwable t) {
-                if (promiseAggregator == null) {
-                    promise.tryFailure(t);
-                } else {
-                    promiseAggregator.setFailure(t);
+                endStream = msg instanceof FullHttpMessage && !((FullHttpMessage) msg).content().isReadable();
+                encoder.writeHeaders(ctx, currentStreamId, http2Headers, 0, endStream, promiseAggregator.newPromise());
+            }
+
+            if (!endStream && msg instanceof HttpContent) {
+                boolean isLastContent = false;
+                Http2Headers trailers = EmptyHttp2Headers.INSTANCE;
+                if (msg instanceof LastHttpContent) {
+                    isLastContent = true;
+
+                    // Convert any trailing headers.
+                    final LastHttpContent lastContent = (LastHttpContent) msg;
+                    trailers = HttpUtil.toHttp2Headers(lastContent.trailingHeaders());
                 }
-            } finally {
-                if (httpMsgNeedRelease) {
-                    httpMsg.release();
+
+                // Write the data
+                final ByteBuf content = ((HttpContent) msg).content();
+                endStream = isLastContent && trailers.isEmpty();
+                release = false;
+                encoder.writeData(ctx, currentStreamId, content, 0, endStream, promiseAggregator.newPromise());
+
+                if (!trailers.isEmpty()) {
+                    // Write trailing headers.
+                    encoder.writeHeaders(ctx, currentStreamId, trailers, 0, true, promiseAggregator.newPromise());
                 }
             }
-        } else {
-            ctx.write(msg, promise);
+
+            promiseAggregator.doneAllocatingPromises();
+        } catch (Throwable t) {
+            promiseAggregator.setFailure(t);
+        } finally {
+            if (release) {
+                ReferenceCountUtil.release(msg);
+            }
         }
     }
 }
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpUtil.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpUtil.java
index 066f3032a8b..61c7bbb9dbc 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpUtil.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/HttpUtil.java
@@ -29,6 +29,7 @@
 import io.netty.handler.codec.http.HttpHeaderUtil;
 import io.netty.handler.codec.http.HttpHeaderValues;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMessage;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
@@ -265,7 +266,7 @@ public static void addHttp2ToHttpHeaders(int streamId, Http2Headers sourceHeader
     /**
      * Converts the given HTTP/1.x headers into HTTP/2 headers.
      */
-    public static Http2Headers toHttp2Headers(FullHttpMessage in) throws Exception {
+    public static Http2Headers toHttp2Headers(HttpMessage in) throws Exception {
         final Http2Headers out = new DefaultHttp2Headers();
         HttpHeaders inHeaders = in.headers();
         if (in instanceof HttpRequest) {
@@ -304,6 +305,16 @@ public static Http2Headers toHttp2Headers(FullHttpMessage in) throws Exception {
         }
 
         // Add the HTTP headers which have not been consumed above
+        return out.add(toHttp2Headers(inHeaders));
+    }
+
+    public static Http2Headers toHttp2Headers(HttpHeaders inHeaders) throws Exception {
+        if (inHeaders.isEmpty()) {
+            return EmptyHttp2Headers.INSTANCE;
+        }
+
+        final Http2Headers out = new DefaultHttp2Headers();
+
         inHeaders.forEachEntry(new EntryVisitor() {
             @Override
             public boolean visit(Entry<CharSequence, CharSequence> entry) throws Exception {


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandlerTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandlerTest.java
index 04d5f7a7434..9960de21e7d 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandlerTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/HttpToHttp2ConnectionHandlerTest.java
@@ -45,9 +45,14 @@
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.DefaultFullHttpRequest;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.DefaultHttpRequest;
+import io.netty.handler.codec.http.DefaultLastHttpContent;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.codec.http2.Http2TestUtil.FrameCountDown;
 import io.netty.util.NetUtil;
 import io.netty.util.concurrent.Future;
@@ -84,6 +89,7 @@ public class HttpToHttp2ConnectionHandlerTest {
     private Channel clientChannel;
     private CountDownLatch requestLatch;
     private CountDownLatch serverSettingsAckLatch;
+    private CountDownLatch trailersLatch;
     private FrameCountDown serverFrameCountDown;
 
     @Before
@@ -104,7 +110,7 @@ public void teardown() throws Exception {
 
     @Test
     public void testJustHeadersRequest() throws Exception {
-        bootstrapEnv(2, 1);
+        bootstrapEnv(2, 1, 0);
         final FullHttpRequest request = new DefaultFullHttpRequest(HTTP_1_1, GET, "/example");
         final HttpHeaders httpHeaders = request.headers();
         httpHeaders.setInt(HttpUtil.ExtensionHeaderNames.STREAM_ID.text(), 5);
@@ -146,7 +152,7 @@ public Void answer(InvocationOnMock in) throws Throwable {
             }
         }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3),
                 any(ByteBuf.class), eq(0), eq(true));
-        bootstrapEnv(3, 1);
+        bootstrapEnv(3, 1, 0);
         final FullHttpRequest request = new DefaultFullHttpRequest(HTTP_1_1, POST, "/example",
                 Unpooled.copiedBuffer(text, UTF_8));
         final HttpHeaders httpHeaders = request.headers();
@@ -175,9 +181,127 @@ public Void answer(InvocationOnMock in) throws Throwable {
         assertEquals(text, receivedBuffers.get(0));
     }
 
-    private void bootstrapEnv(int requestCountDown, int serverSettingsAckCount) throws Exception {
+    @Test
+    public void testRequestWithBodyAndTrailingHeaders() throws Exception {
+        final String text = "foooooogoooo";
+        final List<String> receivedBuffers = Collections.synchronizedList(new ArrayList<String>());
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock in) throws Throwable {
+                receivedBuffers.add(((ByteBuf) in.getArguments()[2]).toString(UTF_8));
+                return null;
+            }
+        }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3),
+                any(ByteBuf.class), eq(0), eq(false));
+        bootstrapEnv(4, 1, 1);
+        final FullHttpRequest request = new DefaultFullHttpRequest(HTTP_1_1, POST, "/example",
+                Unpooled.copiedBuffer(text, UTF_8));
+        final HttpHeaders httpHeaders = request.headers();
+        httpHeaders.set(HttpHeaderNames.HOST, "http://your_user-name123@www.example.org:5555/example");
+        httpHeaders.add("foo", "goo");
+        httpHeaders.add("foo", "goo2");
+        httpHeaders.add("foo2", "goo2");
+        final Http2Headers http2Headers =
+                new DefaultHttp2Headers().method(as("POST")).path(as("/example"))
+                        .authority(as("www.example.org:5555")).scheme(as("http"))
+                        .add(as("foo"), as("goo")).add(as("foo"), as("goo2"))
+                        .add(as("foo2"), as("goo2"));
+
+        request.trailingHeaders().add("trailing", "bar");
+
+        final Http2Headers http2TrailingHeaders = new DefaultHttp2Headers().add(as("trailing"), as("bar"));
+
+        ChannelPromise writePromise = newPromise();
+        ChannelFuture writeFuture = clientChannel.writeAndFlush(request, writePromise);
+
+        assertTrue(writePromise.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(writePromise.isSuccess());
+        assertTrue(writeFuture.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(writeFuture.isSuccess());
+        awaitRequests();
+        verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(http2Headers), eq(0),
+                anyShort(), anyBoolean(), eq(0), eq(false));
+        verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), any(ByteBuf.class), eq(0),
+                eq(false));
+        verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(http2TrailingHeaders), eq(0),
+                anyShort(), anyBoolean(), eq(0), eq(true));
+        assertEquals(1, receivedBuffers.size());
+        assertEquals(text, receivedBuffers.get(0));
+    }
+
+    @Test
+    public void testChunkedRequestWithBodyAndTrailingHeaders() throws Exception {
+        final String text = "foooooo";
+        final String text2 = "goooo";
+        final List<String> receivedBuffers = Collections.synchronizedList(new ArrayList<String>());
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock in) throws Throwable {
+                receivedBuffers.add(((ByteBuf) in.getArguments()[2]).toString(UTF_8));
+                return null;
+            }
+        }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3),
+                any(ByteBuf.class), eq(0), eq(false));
+        bootstrapEnv(4, 1, 1);
+        final HttpRequest request = new DefaultHttpRequest(HTTP_1_1, POST, "/example");
+        final HttpHeaders httpHeaders = request.headers();
+        httpHeaders.set(HttpHeaderNames.HOST, "http://your_user-name123@www.example.org:5555/example");
+        httpHeaders.add(HttpHeaderNames.TRANSFER_ENCODING, "chunked");
+        httpHeaders.add("foo", "goo");
+        httpHeaders.add("foo", "goo2");
+        httpHeaders.add("foo2", "goo2");
+        final Http2Headers http2Headers =
+                new DefaultHttp2Headers().method(as("POST")).path(as("/example"))
+                        .authority(as("www.example.org:5555")).scheme(as("http"))
+                        .add(as("foo"), as("goo")).add(as("foo"), as("goo2"))
+                        .add(as("foo2"), as("goo2"));
+
+        final DefaultHttpContent httpContent = new DefaultHttpContent(Unpooled.copiedBuffer(text, UTF_8));
+        final LastHttpContent lastHttpContent = new DefaultLastHttpContent(Unpooled.copiedBuffer(text2, UTF_8));
+
+        lastHttpContent.trailingHeaders().add("trailing", "bar");
+
+        final Http2Headers http2TrailingHeaders = new DefaultHttp2Headers().add(as("trailing"), as("bar"));
+
+        ChannelPromise writePromise = newPromise();
+        ChannelFuture writeFuture = clientChannel.write(request, writePromise);
+        ChannelPromise contentPromise = newPromise();
+        ChannelFuture contentFuture = clientChannel.write(httpContent, contentPromise);
+        ChannelPromise lastContentPromise = newPromise();
+        ChannelFuture lastContentFuture = clientChannel.write(lastHttpContent, lastContentPromise);
+
+        clientChannel.flush();
+
+        assertTrue(writePromise.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(writePromise.isSuccess());
+        assertTrue(writeFuture.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(writeFuture.isSuccess());
+
+        assertTrue(contentPromise.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(contentPromise.isSuccess());
+        assertTrue(contentFuture.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(contentFuture.isSuccess());
+
+        assertTrue(lastContentPromise.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(lastContentPromise.isSuccess());
+        assertTrue(lastContentFuture.awaitUninterruptibly(WAIT_TIME_SECONDS, SECONDS));
+        assertTrue(lastContentFuture.isSuccess());
+
+        awaitRequests();
+        verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(http2Headers), eq(0),
+                anyShort(), anyBoolean(), eq(0), eq(false));
+        verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), any(ByteBuf.class), eq(0),
+                eq(false));
+        verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(http2TrailingHeaders), eq(0),
+                anyShort(), anyBoolean(), eq(0), eq(true));
+        assertEquals(1, receivedBuffers.size());
+        assertEquals(text + text2, receivedBuffers.get(0));
+    }
+
+    private void bootstrapEnv(int requestCountDown, int serverSettingsAckCount, int trailersCount) throws Exception {
         requestLatch = new CountDownLatch(requestCountDown);
         serverSettingsAckLatch = new CountDownLatch(serverSettingsAckCount);
+        trailersLatch = trailersCount == 0 ? null : new CountDownLatch(trailersCount);
 
         sb = new ServerBootstrap();
         cb = new Bootstrap();
@@ -188,7 +312,8 @@ private void bootstrapEnv(int requestCountDown, int serverSettingsAckCount) thro
             @Override
             protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
-                serverFrameCountDown = new FrameCountDown(serverListener, serverSettingsAckLatch, requestLatch);
+                serverFrameCountDown =
+                        new FrameCountDown(serverListener, serverSettingsAckLatch, requestLatch, null, trailersLatch);
                 p.addLast(new HttpToHttp2ConnectionHandler(true, serverFrameCountDown));
             }
         });
@@ -213,6 +338,10 @@ protected void initChannel(Channel ch) throws Exception {
 
     private void awaitRequests() throws Exception {
         assertTrue(requestLatch.await(WAIT_TIME_SECONDS, SECONDS));
+        if (trailersLatch != null) {
+            assertTrue(trailersLatch.await(WAIT_TIME_SECONDS, SECONDS));
+        }
+        assertTrue(serverSettingsAckLatch.await(WAIT_TIME_SECONDS, SECONDS));
     }
 
     private ChannelHandlerContext ctx() {