Make the inbound queue in DefaultHttp2Stream use the Buffer type consistently

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/http2/DefaultHttp2Stream.java

@@ -40,10 +40,10 @@
  */
 abstract class DefaultHttp2Stream<S extends DefaultHttp2Stream<S>> implements HttpStream, Http2Stream {
 
-  private static final HttpHeaders EMPTY = new HttpHeaders(EmptyHttp2Headers.INSTANCE);
+  private static final Buffer EMPTY = BufferInternal.buffer(Unpooled.EMPTY_BUFFER);
 
   private final OutboundMessageQueue<MessageWrite> outboundQueue;
-  private final InboundMessageQueue<Object> inboundQueue;
+  private final InboundMessageQueue<Buffer> inboundQueue;
   private final Http2Connection connection;
   protected final VertxInternal vertx;
   protected final ContextInternal context;
@@ -56,6 +56,9 @@ abstract class DefaultHttp2Stream<S extends DefaultHttp2Stream<S>> implements Ht
   private boolean trailersSent;
   private boolean writable;
 
+  // Written by event-loop - read by context with happens-before relationship
+  private MultiMap trailers;
+
   // Client context
   private StreamPriority priority;
   private long bytesRead;
@@ -85,14 +88,13 @@ abstract class DefaultHttp2Stream<S extends DefaultHttp2Stream<S>> implements Ht
     this.id = id_;
     this.inboundQueue = new InboundMessageQueue<>(connection.context().eventLoop(), context.executor()) {
       @Override
-      protected void handleMessage(Object item) {
-        if (item instanceof MultiMap) {
-          handleTrailers((MultiMap) item);
+      protected void handleMessage(Buffer item) {
+        if (item == EMPTY) {
+          handleTrailers(trailers);
         } else {
-          Buffer data = (Buffer) item;
-          int len = data.length();
+          int len = item.length();
           connection.context().execute(len, v -> connection.consumeCredits(DefaultHttp2Stream.this.id, v));
-          handleData(data);
+          handleData(item);
         }
       }
     };
@@ -241,20 +243,21 @@ public void onWritabilityChanged() {
   }
 
   public final void onTrailers() {
-    onTrailers(EMPTY);
+    onTrailers(null);
   }
 
-  public final void onTrailers(HttpHeaders trailers) {
+  public final void onTrailers(HttpHeaders t) {
     if (trailersReceived) {
       throw new IllegalStateException();
     }
     trailersReceived = true;
+    trailers = t;
     StreamObserver observer = observer();
     if (observer != null) {
       observer.observeInboundTrailers(bytesRead);
     }
     connection.flushBytesRead();
-    inboundQueue.write(trailers);
+    inboundQueue.write(EMPTY);
   }
 
   public final long id() {