Rewrite the RecordParser stream handling

eclipse-vertx · Oct 23, 2018 · 863eaa3 · 863eaa3
1 parent 8035628
commit 863eaa3
Show file tree

Hide file tree

Showing 2 changed files with 61 additions and 37 deletions.
diff --git a/src/main/java/io/vertx/core/parsetools/impl/RecordParserImpl.java b/src/main/java/io/vertx/core/parsetools/impl/RecordParserImpl.java
@@ -15,7 +15,6 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.Arguments;
 import io.vertx.core.parsetools.RecordParser;
-import io.vertx.core.queue.Queue;
 import io.vertx.core.streams.ReadStream;
 
 import java.util.Objects;
@@ -30,29 +29,26 @@ public class RecordParserImpl implements RecordParser {
   private int pos;            // Current position in buffer
   private int start;          // Position of beginning of current record
   private int delimPos;       // Position of current match in delimiter array
+  private int next = -1;      // Position of the next matching record
 
   private boolean delimited;
   private byte[] delim;
   private int recordSize;
   private int maxRecordSize;
+  private long demand = Long.MAX_VALUE;
+  private Handler<Buffer> eventHandler;
   private Handler<Void> endHandler;
   private Handler<Throwable> exceptionHandler;
 
   private final ReadStream<Buffer> stream;
-  private final Queue<Buffer> pending;
 
   private RecordParserImpl(ReadStream<Buffer> stream) {
     this.stream = stream;
-    this.pending = Queue
-      .<Buffer>queue()
-      .writableHandler(v -> {
-        stream.resume();
-      });
   }
 
   public void setOutput(Handler<Buffer> output) {
     Objects.requireNonNull(output, "output");
-    this.pending.handler(output);
+    eventHandler = output;
   }
 
   /**
@@ -175,15 +171,38 @@ public RecordParser maxRecordSize(int size) {
   private void handleParsing() {
     int len = buff.length();
     do {
-      Buffer event;
+      if (next == -1) {
-      if (delimited) {
+        if (delimited) {
-        event = parseDelimited();
+          next = parseDelimited();
-      } else {
+        } else {
-        event = parseFixed();
+          next = parseFixed();
+        }
       }
-      if (event != null) {
+      if (demand > 0L) {
-        handleEvent(event);
+        if (next == -1) {
+          ReadStream<Buffer> s = stream;
+          if (s != null) {
+            s.resume();
+          }
+          break;
+        }
+        if (demand != Long.MAX_VALUE) {
+          demand--;
+        }
+        Buffer event = buff.getBuffer(start, next);
+        start = pos;
+        next = -1;
+        Handler<Buffer> handler = eventHandler;
+        if (handler != null) {
+          handler.handle(event);
+        }
       } else {
+        if (next != -1) {
+          ReadStream<Buffer> s = stream;
+          if (s != null) {
+            s.pause();
+          }
+        }
         break;
       }
     } while (true);
@@ -199,23 +218,15 @@ private void handleParsing() {
     start = 0;
   }
 
-  private void handleEvent(Buffer event) {
+  private int parseDelimited() {
-    if (!pending.add(event) && stream != null) {
-      stream.pause();
-    }
-  }
-
-  private Buffer parseDelimited() {
     int len = buff.length();
     for (; pos < len; pos++) {
       if (buff.getByte(pos) == delim[delimPos]) {
         delimPos++;
         if (delimPos == delim.length) {
           pos++;
-          Buffer ret = buff.getBuffer(start, pos - delim.length);
-          start = pos;
           delimPos = 0;
-          return ret;
+          return pos - delim.length;
         }
       } else {
         if (delimPos > 0) {
@@ -224,19 +235,17 @@ private Buffer parseDelimited() {
         }
       }
     }
-    return null;
+    return -1;
   }
 
-  private Buffer parseFixed() {
+  private int parseFixed() {
     int len = buff.length();
     if (len - start >= recordSize) {
       int end = start + recordSize;
-      Buffer ret = buff.getBuffer(start, end);
+      pos = end;
-      start = end;
+      return end;
-      pos = start;
-      return ret;
     }
-    return null;
+    return -1;
   }
 
   /**
@@ -276,7 +285,7 @@ public RecordParser exceptionHandler(Handler<Throwable> handler) {
 
   @Override
   public RecordParser handler(Handler<Buffer> handler) {
-    pending.handler(handler);
+    eventHandler = handler;
     if (stream != null) {
       if (handler != null) {
         stream.endHandler(v -> end());
@@ -297,20 +306,24 @@ public RecordParser handler(Handler<Buffer> handler) {
 
   @Override
   public RecordParser pause() {
-    pending.pause();
+    demand = 0L;
     return this;
   }
 
   @Override
   public RecordParser fetch(long amount) {
-    pending.take(amount);
+    Arguments.require(amount > 0, "Fetch amount must be > 0");
+    demand += amount;
+    if (demand < 0L) {
+      demand = Long.MAX_VALUE;
+    }
+    handleParsing();
     return this;
   }
 
   @Override
   public RecordParser resume() {
-    pending.resume();
+    return fetch(Long.MAX_VALUE);
-    return this;
   }
 
   @Override

diff --git a/src/test/java/io/vertx/core/parsetools/RecordParserTest.java b/src/test/java/io/vertx/core/parsetools/RecordParserTest.java
@@ -333,11 +333,22 @@ public void testWrapReadStream() {
       assertEquals("item-" + i, records.poll());
     }
     assertNull(records.poll());
+    assertFalse(stream.isPaused());
 
     stream.end();
     assertEquals(1, ends.get());
   }
 
+  @Test
+  public void testPausedStreamShouldNotPauseOnIncompleteMatch() {
+    FakeStream stream = new FakeStream();
+    RecordParser parser = RecordParser.newDelimited("\r\n", stream);
+    parser.handler(event -> {});
+    parser.pause();
+    stream.handle("abc");
+    assertFalse(stream.isPaused());
+  }
+
   private void doTestDelimitedMaxRecordSize(final Buffer input, Buffer delim, Integer[] chunkSizes, int maxRecordSize,
                                             Handler<Throwable> exHandler, final Buffer... expected) {
     final Buffer[] results = new Buffer[expected.length];