Rewrite the Queue, rename to InboundBuffer and make it more internal

eclipse-vertx · Oct 23, 2018 · 38cc7f5 · 38cc7f5
1 parent c3a6f75
commit 38cc7f5
Show file tree

Hide file tree

Showing 14 changed files with 1,093 additions and 1,361 deletions.
diff --git a/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java b/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java
@@ -24,7 +24,7 @@
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.queue.Queue;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -68,7 +68,7 @@ public class AsyncFileImpl implements AsyncFile {
   private int maxWrites = 128 * 1024;    // TODO - we should tune this for best performance
   private int lwm = maxWrites / 2;
   private int readBufferSize = DEFAULT_READ_BUFFER_SIZE;
-  private Queue<Buffer> queue;
+  private InboundBuffer<Buffer> queue;
   private Handler<Void> endHandler;
   private long readPos;
 
@@ -100,9 +100,9 @@ public class AsyncFileImpl implements AsyncFile {
       throw new FileSystemException(e);
     }
     this.context = context;
-    this.queue = Queue.queue(context, 0);
+    this.queue = new InboundBuffer<>(context, 0);
 
-    queue.writableHandler(v -> {
+    queue.drainHandler(v -> {
       doRead();
     });
   }
@@ -137,7 +137,7 @@ public synchronized AsyncFile read(Buffer buffer, int offset, long position, int
 
   @Override
   public AsyncFile fetch(long amount) {
-    queue.take(amount);
+    queue.fetch(amount);
     return this;
   }
 
@@ -338,7 +338,7 @@ private synchronized void doRead() {
           handleEnd();
         } else {
           readPos += buffer.length();
-          if (queue.add(buffer)) {
+          if (queue.write(buffer)) {
             doRead();
           }
         }

diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
@@ -39,7 +39,7 @@
 import io.vertx.core.net.impl.NetSocketImpl;
 import io.vertx.core.net.impl.VertxNetHandler;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
-import io.vertx.core.queue.Queue;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 import java.net.URI;
 import java.util.*;
@@ -195,15 +195,15 @@ private static class StreamImpl implements HttpClientStream {
     private boolean requestEnded;
     private boolean responseEnded;
     private boolean reset;
-    private Queue<Buffer> queue;
+    private InboundBuffer<Buffer> queue;
     private MultiMap trailers;
     private StreamImpl next;
 
     StreamImpl(Http1xClientConnection conn, int id, Handler<AsyncResult<HttpClientStream>> handler) {
       this.conn = conn;
       this.fut = Future.<HttpClientStream>future().setHandler(handler);
       this.id = id;
-      this.queue = Queue.queue(conn.context, 0);
+      this.queue = new InboundBuffer<>(conn.context, 5);
     }
 
     private void append(StreamImpl s) {
@@ -304,7 +304,7 @@ private void sendRequest(
     }
 
     private boolean handleChunk(Buffer buff) {
-      return queue.add(buff);
+      return queue.write(buff);
     }
 
     @Override
@@ -342,7 +342,7 @@ public void doPause() {
 
     @Override
     public void doFetch(long amount) {
-      queue.take(amount);
+      queue.fetch(amount);
     }
 
     @Override
@@ -454,7 +454,7 @@ private HttpClientResponseImpl beginResponse(HttpResponse resp) {
           response.handleEnd(trailers);
         }
       });
-      queue.writableHandler(v -> {
+      queue.drainHandler(v -> {
         if (!responseEnded) {
           conn.doResume();
         }

diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerFileUploadImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerFileUploadImpl.java
@@ -13,15 +13,13 @@
 
 import io.vertx.core.Context;
 import io.vertx.core.Handler;
-import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.file.AsyncFile;
 import io.vertx.core.file.OpenOptions;
 import io.vertx.core.http.HttpServerFileUpload;
 import io.vertx.core.http.HttpServerRequest;
-import io.vertx.core.queue.Queue;
 import io.vertx.core.streams.Pump;
-import io.vertx.core.streams.ReadStream;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 import java.nio.charset.Charset;
 
@@ -49,7 +47,7 @@ class HttpServerFileUploadImpl implements HttpServerFileUpload {
   private Handler<Throwable> exceptionHandler;
 
   private long size;
-  private Queue<Buffer> pending;
+  private InboundBuffer<Buffer> pending;
   private boolean complete;
   private boolean lazyCalculateSize;
 
@@ -64,7 +62,7 @@ class HttpServerFileUploadImpl implements HttpServerFileUpload {
     this.contentTransferEncoding = contentTransferEncoding;
     this.charset = charset;
     this.size = size;
-    this.pending = Queue.<Buffer>queue(context)
+    this.pending = new InboundBuffer<Buffer>(context)
       .emptyHandler(v -> {
         if (complete) {
           handleComplete();
@@ -173,7 +171,7 @@ synchronized void receiveData(Buffer data) {
   }
 
   synchronized void doReceiveData(Buffer data) {
-    if (!pending.add(data)) {
+    if (!pending.write(data)) {
       req.pause();
     }
   }

diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java
@@ -29,7 +29,7 @@
 import io.vertx.core.net.NetSocket;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.metrics.Metrics;
-import io.vertx.core.queue.Queue;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.net.ssl.SSLSession;
@@ -85,7 +85,7 @@ public class HttpServerRequestImpl implements HttpServerRequest {
   private long bytesRead;
 
   private boolean paused;
-  private Queue<Buffer> pending;
+  private InboundBuffer<Buffer> pending;
 
   HttpServerRequestImpl(Http1xServerConnection conn, DefaultHttpRequest request) {
     this.conn = conn;
@@ -104,10 +104,10 @@ void setRequest(DefaultHttpRequest request) {
     }
   }
 
-  private Queue<Buffer> pendingQueue() {
+  private InboundBuffer<Buffer> pendingQueue() {
     if (pending == null) {
-      pending = Queue.queue(conn.getContext(), 8);
+      pending = new InboundBuffer<>(conn.getContext(), 8);
-      pending.writableHandler(v -> conn.doResume());
+      pending.drainHandler(v -> conn.doResume());
       pending.emptyHandler(v -> {
         if (ended) {
           doEnd();
@@ -120,7 +120,7 @@ private Queue<Buffer> pendingQueue() {
 
   private void enqueueData(Buffer chunk) {
     // We queue requests if paused or a request is in progress to prevent responses being written in the wrong order
-    if (!pendingQueue().add(chunk)) {
+    if (!pendingQueue().write(chunk)) {
       // We only pause when we are actively called by the connection
       conn.doPause();
     }
@@ -327,7 +327,7 @@ public HttpServerRequest pause() {
   @Override
   public HttpServerRequest fetch(long amount) {
     synchronized (conn) {
-      pendingQueue().take(amount);
+      pendingQueue().fetch(amount);
       return this;
     }
   }

diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java
@@ -20,7 +20,7 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.VertxInternal;
-import io.vertx.core.queue.Queue;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
@@ -35,7 +35,7 @@ abstract class VertxHttp2Stream<C extends Http2ConnectionBase> {
   protected final ChannelHandlerContext handlerContext;
   protected final Http2Stream stream;
 
-  private Queue<Buffer> pending;
+  private InboundBuffer<Buffer> pending;
   private int pendingBytes;
   private MultiMap trailers;
   private boolean writable;
@@ -47,15 +47,16 @@ abstract class VertxHttp2Stream<C extends Http2ConnectionBase> {
     this.stream = stream;
     this.context = conn.getContext();
     this.writable = writable;
-    this.pending = Queue.queue(context, 5);
+    this.pending = new InboundBuffer<>(context, 5);
 
-    pending.writableHandler(v -> {
+    pending.drainHandler(v -> {
       int numBytes = pendingBytes;
       pendingBytes = 0;
       conn.handler.consume(stream, numBytes);
     });
 
     pending.handler(this::handleData);
+    pending.exceptionHandler(context.exceptionHandler());
     pending.emptyHandler(v -> {
       if (trailers != null) {
         handleEnd(trailers);
@@ -70,7 +71,7 @@ void onResetRead(long code) {
   }
 
   boolean onDataRead(Buffer data) {
-    boolean read = pending.add(data);
+    boolean read = pending.write(data);
     if (!read) {
       pendingBytes += data.length();
     }
@@ -110,7 +111,7 @@ public void doResume() {
   }
 
   public void doFetch(long amount) {
-    pending.take(amount);
+    pending.fetch(amount);
   }
 
   boolean isNotWritable() {

diff --git a/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java b/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java
@@ -23,7 +23,7 @@
 import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.net.SocketAddress;
-import io.vertx.core.queue.Queue;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.net.ssl.SSLSession;
@@ -46,7 +46,7 @@ public abstract class WebSocketImplBase<S extends WebSocketBase> implements WebS
   private final String binaryHandlerID;
   private final int maxWebSocketFrameSize;
   private final int maxWebSocketMessageSize;
-  private final Queue<Buffer> pending;
+  private final InboundBuffer<Buffer> pending;
   private MessageConsumer binaryHandlerRegistration;
   private MessageConsumer textHandlerRegistration;
   private String subProtocol;
@@ -69,9 +69,9 @@ public abstract class WebSocketImplBase<S extends WebSocketBase> implements WebS
     this.conn = conn;
     this.maxWebSocketFrameSize = maxWebSocketFrameSize;
     this.maxWebSocketMessageSize = maxWebSocketMessageSize;
-    this.pending = Queue.<Buffer>queue(conn.getContext());
+    this.pending = new InboundBuffer<>(conn.getContext());
 
-    pending.writableHandler(v -> {
+    pending.drainHandler(v -> {
       conn.doResume();
     });
   }
@@ -284,7 +284,7 @@ void handleFrame(WebSocketFrameInternal frame) {
     synchronized (conn) {
       if (frame.type() != FrameType.CLOSE) {
         conn.reportBytesRead(frame.length());
-        if (!pending.add(frame.binaryData())) {
+        if (!pending.write(frame.binaryData())) {
           conn.doPause();
         }
       }
@@ -557,7 +557,7 @@ public S resume() {
   @Override
   public S fetch(long amount) {
     if (!isClosed()) {
-      pending.take(amount);
+      pending.fetch(amount);
     }
     return (S) this;
   }

diff --git a/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java b/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java
@@ -36,7 +36,7 @@
 import io.vertx.core.net.NetSocket;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.metrics.TCPMetrics;
-import io.vertx.core.queue.Queue;
+import io.vertx.core.streams.impl.InboundBuffer;
 
 import java.io.File;
 import java.io.IOException;
@@ -71,7 +71,7 @@ public class NetSocketImpl extends ConnectionBase implements NetSocketInternal {
   private final TCPMetrics metrics;
   private Handler<Void> endHandler;
   private Handler<Void> drainHandler;
-  private Queue<Object> pending;
+  private InboundBuffer<Object> pending;
   private MessageConsumer registration;
   private boolean closed;
 
@@ -87,8 +87,8 @@ public NetSocketImpl(VertxInternal vertx, ChannelHandlerContext channel, SocketA
     this.writeHandlerID = "__vertx.net." + UUID.randomUUID().toString();
     this.remoteAddress = remoteAddress;
     this.metrics = metrics;
-    pending = Queue.queue(context, 0);
+    pending = new InboundBuffer<>(context);
-    pending.writableHandler(v -> doResume());
+    pending.drainHandler(v -> doResume());
     pending.handler(NULL_MSG_HANDLER);
   }
 
@@ -188,7 +188,7 @@ public synchronized NetSocket pause() {
 
   @Override
   public NetSocket fetch(long amount) {
-    pending.take(amount);
+    pending.fetch(amount);
     return this;
   }
 
@@ -361,7 +361,7 @@ protected void handleClosed() {
 
   public synchronized void handleMessage(Object msg) {
     checkContext();
-    if (!pending.add(msg)) {
+    if (!pending.write(msg)) {
       doPause();
     }
   }