Prototype HttpBodyDecoder

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/HttpBodyDecoder.java

@@ -0,0 +1,13 @@
+package io.vertx.core.http.impl;
+
+import io.netty.buffer.ByteBuf;
+
+public interface HttpBodyDecoder {
+
+  int handle(ByteBuf content);
+
+  void next();
+
+  void end();
+
+}

vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientStream.java

@@ -46,5 +46,8 @@ public interface HttpClientStream extends HttpStream {
   HttpClientStream fetch(long amount);
 
   HttpClientStream updatePriority(StreamPriority streamPriority);
-
+  @Override
+  default HttpClientStream bodyDecoder(HttpBodyDecoder decoder) {
+    return (HttpClientStream)HttpStream.super.bodyDecoder(decoder);
+  }
 }

vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerStream.java

@@ -54,7 +54,9 @@ public interface HttpServerStream extends HttpStream {
 
   void sendFile(ChunkedInput<ByteBuf> file, Promise<Void> promise);
 
-
   HttpServerStream updatePriority(StreamPriority streamPriority);
-
+  @Override
+  default HttpServerStream bodyDecoder(HttpBodyDecoder decoder) {
+    return (HttpServerStream)HttpStream.super.bodyDecoder(decoder);
+  }
 }

vertx-core/src/main/java/io/vertx/core/http/impl/HttpStream.java

@@ -15,10 +15,7 @@
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.HttpConnection;
-import io.vertx.core.http.HttpFrame;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.StreamPriority;
+import io.vertx.core.http.*;
 import io.vertx.core.internal.ContextInternal;
 
 /**
@@ -49,6 +46,10 @@ public interface HttpStream {
 
   Future<Boolean> cancel();
 
+  default HttpStream bodyDecoder(HttpBodyDecoder decoder) {
+    throw new UnsupportedOperationException();
+  }
+
   HttpStream resetHandler(Handler<Long> handler);
   HttpStream exceptionHandler(Handler<Throwable> handler);
   HttpStream customFrameHandler(Handler<HttpFrame> handler);

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

@@ -15,7 +15,6 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.EventLoop;
-import io.netty.handler.codec.http2.EmptyHttp2Headers;
 import io.netty.handler.stream.ChunkedInput;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
@@ -28,22 +27,24 @@
 import io.vertx.core.http.impl.HttpUtils;
 import io.vertx.core.http.impl.headers.HttpHeaders;
 import io.vertx.core.http.impl.observability.StreamObserver;
+import io.vertx.core.http.impl.HttpBodyDecoder;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.VertxInternal;
 import io.vertx.core.internal.buffer.BufferInternal;
 import io.vertx.core.internal.concurrent.InboundMessageQueue;
 import io.vertx.core.internal.concurrent.OutboundMessageQueue;
 import io.vertx.core.net.impl.MessageWrite;
+import io.vertx.core.net.impl.VertxHandler;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
 abstract class DefaultHttp2Stream<S extends DefaultHttp2Stream<S>> implements HttpStream, Http2Stream {
 
-  private static final Buffer EMPTY = BufferInternal.buffer(Unpooled.EMPTY_BUFFER);
+  private static final ByteBuf EMPTY = Unpooled.wrappedBuffer(new byte[1]);
 
   private final OutboundMessageQueue<MessageWrite> outboundQueue;
-  private final InboundMessageQueue<Buffer> inboundQueue;
+  private final InboundMessageQueue<ByteBuf> inboundQueue;
   private final Http2Connection connection;
   protected final VertxInternal vertx;
   protected final ContextInternal context;
@@ -77,6 +78,9 @@ abstract class DefaultHttp2Stream<S extends DefaultHttp2Stream<S>> implements Ht
   private Handler<StreamPriority> priorityChangeHandler;
   private Handler<Void> drainHandler;
 
+  //
+  private HttpBodyDecoder decoder;
+
   DefaultHttp2Stream(Http2Connection connection, ContextInternal context) {
     this(-1, connection, context, true);
   }
@@ -88,13 +92,23 @@ 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(Buffer item) {
-        if (item == EMPTY) {
-          handleTrailers(trailers);
+      protected long evalMessage(ByteBuf data) {
+        if (data == EMPTY) {
+          return 1;
         } else {
-          int len = item.length();
+          int len = data.readableBytes();
           connection.context().execute(len, v -> connection.consumeCredits(DefaultHttp2Stream.this.id, v));
-          handleData(item);
+          return decoder.handle(data);
+        }
+      }
+      @Override
+      protected void handleMessage(ByteBuf data, long amount) {
+        if (data == EMPTY) {
+          handleTrailers(trailers);
+        } else {
+          for (int i = 0;i < amount;i++) {
+            decoder.next();
+          }
         }
       }
     };
@@ -134,6 +148,26 @@ protected void handleDrained() {
         this.trailersReceived = true;
       }
     }
+    this.decoder = new HttpBodyDecoder() {
+      private Buffer buffer;
+      @Override
+      public int handle(ByteBuf content) {
+        buffer = BufferInternal.buffer(VertxHandler.safeBuffer(content));
+        return 1;
+      }
+      @Override
+      public void next() {
+        Buffer item = buffer;
+        buffer = null;
+        Handler<Buffer> handler = dataHandler;
+        if (handler != null) {
+          context.dispatch(item, handler);
+        }
+      }
+      @Override
+      public void end() {
+      }
+    };
   }
 
   public final HttpVersion version() {
@@ -230,8 +264,12 @@ public void onCustomFrame(int type, int flags, Buffer payload) {
     context.execute(new HttpFrameImpl(type, flags, payload), this::handleCustomFrame);
   }
 
-  public void onData(Buffer data) {
-    bytesRead += data.length();
+  public void onData(ByteBuf data) {
+    // Warning : we retain the buffer and we don't release it on close
+    // when close happens we should flush the queue and cumulate the result
+    // for later delivery
+    data.retain();
+    bytesRead += data.readableBytes();
     inboundQueue.write(data);
   }
 
@@ -426,6 +464,12 @@ void writeData0(ByteBuf buf, boolean end, Promise<Void> promise) {
     connection.writeData(id, chunk, end, promise);
   }
 
+  @Override
+  public S bodyDecoder(HttpBodyDecoder decoder) {
+    this.decoder = decoder;
+    return (S)this;
+  }
+
   @Override
   public Future<Boolean> cancel() {
     return writeReset(0x08L).map(true);
@@ -481,13 +525,6 @@ public S dataHandler(Handler<Buffer> handler) {
     return (S)this;
   }
 
-  private void handleData(Buffer buf) {
-    Handler<Buffer> handler = dataHandler;
-    if (handler != null) {
-      context.dispatch(buf, handler);
-    }
-  }
-
   public S customFrameHandler(Handler<HttpFrame> handler) {
     customFrameHandler = handler;
     return (S)this;
@@ -510,6 +547,7 @@ private void handleTrailers(MultiMap trailers) {
     if (handler != null) {
       context.dispatch(trailers, handler);
     }
+    decoder.end();
   }
 
   public S resetHandler(Handler<Long> handler) {

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

@@ -10,6 +10,7 @@
  */
 package io.vertx.core.http.impl.http2;
 
+import io.netty.buffer.ByteBuf;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.GoAway;
 import io.vertx.core.http.StreamPriority;
@@ -36,7 +37,7 @@ public interface Http2Stream {
 
   void onPriorityChange(StreamPriority streamPriority);
   void onHeaders(HttpHeaders headers);
-  void onData(Buffer buffer);
+  void onData(ByteBuf buffer);
   void onCustomFrame(int type, int flags, Buffer payload);
   void onTrailers();
   void onTrailers(HttpHeaders trailers);

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

@@ -53,7 +53,7 @@ abstract class Http2ConnectionImpl extends ConnectionBase implements Http2FrameL
 
   private static final Logger log = LoggerFactory.getLogger(Http2ConnectionImpl.class);
 
-  private static ByteBuf safeBuffer(ByteBuf buf) {
+  public static ByteBuf safeBuffer(ByteBuf buf) {
     ByteBuf buffer = VertxByteBufAllocator.DEFAULT.heapBuffer(buf.readableBytes());
     buffer.writeBytes(buf);
     return buffer;
@@ -300,9 +300,7 @@ public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorC
   public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) {
     Http2Stream stream = stream(streamId);
     if (stream != null) {
-      data = safeBuffer(data);
-      Buffer buff = BufferInternal.buffer(data);
-      stream.onData(buff);
+      stream.onData(data);
       if (endOfStream) {
         stream.onTrailers();
       }

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

@@ -106,15 +106,12 @@ final void registerChannel(S stream, Http2FrameStream frameStream, ChannelHandle
 
   void receiveData(ChannelHandlerContext chctx, int streamId, ByteBuf content, boolean ended, int initialWindowSize) {
     StreamChannel channel = channels.get(streamId);
-    ByteBuf buffer = VertxByteBufAllocator.DEFAULT.heapBuffer(content.readableBytes());
-    buffer.writeBytes(content, content.readerIndex(), content.readableBytes());
-    Buffer buff = BufferInternal.buffer(buffer);
-    channel.bytesConsumed += buff.length();
+    channel.bytesConsumed += content.readableBytes();
     if (channel.bytesConsumed > initialWindowSize) {
       chctx.channel().config().setAutoRead(false);
     }
     S stream = channel.stream;
-    stream.onData(buff);
+    stream.onData(content);
     if (ended) {
       stream.onTrailers();
     }

vertx-core/src/test/java/io/vertx/tests/http/HttpBodyCodecTest.java

@@ -0,0 +1,55 @@
+package io.vertx.tests.http;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.*;
+import io.vertx.core.http.impl.HttpResponseHead;
+import io.vertx.core.http.impl.HttpServerConnection;
+import io.vertx.core.http.impl.headers.HttpResponseHeaders;
+import io.vertx.core.http.impl.HttpBodyDecoder;
+import io.vertx.test.core.TestUtils;
+import io.vertx.test.http.HttpConfigurator;
+import io.vertx.test.http.SimpleHttpTest2;
+import org.junit.Test;
+
+import static org.junit.Assert.fail;
+
+public class HttpBodyCodecTest extends SimpleHttpTest2 {
+
+  public HttpBodyCodecTest() {
+    super(HttpConfigurator.H2.MULTIPLEX);
+  }
+
+  @Test
+  public void testBodyDecoder() throws Exception {
+    Buffer body = Buffer.buffer(TestUtils.randomAlphaString(1024));
+    server.connectionHandler(connection -> {
+      HttpServerConnection serverConnection = (HttpServerConnection) connection;
+      serverConnection.streamHandler(stream -> {
+        stream.bodyDecoder(new HttpBodyDecoder() {
+          @Override
+          public int handle(ByteBuf content) {
+            ReferenceCountUtil.release(content);
+            return 10;
+          }
+          @Override
+          public void next() {
+          }
+          @Override
+          public void end() {
+            stream.writeHead(new HttpResponseHead(200, null, new HttpResponseHeaders(serverConnection.newHeaders())), null, true);
+          }
+        });
+      });
+    });
+    server.requestHandler(request -> {
+      fail();
+    });
+    startServer(testAddress);
+    client.request(new RequestOptions(requestOptions).setMethod(HttpMethod.POST))
+      .compose(request -> {
+        return request.send(body).compose(HttpClientResponse::end);
+      }).await();
+  }
+}