Repository: eclipse-vertx/vert.x

Problem statement:
RecordParser implement ReadStream<Buffer>
That way we can trivially pump files or other ReadStream sources through it.

pump(asyncFile, recordParser) 


Hints:
good idea, would you mind to contribute it ?

Patch:
diff --git a/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java b/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
--- a/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
@@ -314,7 +314,7 @@ private String getClusterPublicHost(EventBusOptions options) {
 
   private Handler<NetSocket> getServerHandler() {
     return socket -> {
-      RecordParser parser = RecordParser.newFixed(4, null);
+      RecordParser parser = RecordParser.newFixed(4);
       Handler<Buffer> handler = new Handler<Buffer>() {
         int size = -1;
 
diff --git a/src/main/java/io/vertx/core/parsetools/RecordParser.java b/src/main/java/io/vertx/core/parsetools/RecordParser.java
--- a/src/main/java/io/vertx/core/parsetools/RecordParser.java
+++ b/src/main/java/io/vertx/core/parsetools/RecordParser.java
@@ -20,6 +20,7 @@
 import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.parsetools.impl.RecordParserImpl;
+import io.vertx.core.streams.ReadStream;
 
 /**
  * A helper class which allows you to easily parse protocols which are delimited by a sequence of bytes, or fixed
@@ -56,7 +57,7 @@
  * @author <a href="mailto:larsdtimm@gmail.com">Lars Timm</a>
  */
 @VertxGen
-public interface RecordParser extends Handler<Buffer> {
+public interface RecordParser extends Handler<Buffer>, ReadStream<Buffer> {
 
   void setOutput(Handler<Buffer> output);
 
@@ -68,7 +69,20 @@ public interface RecordParser extends Handler<Buffer> {
    * @param output  handler that will receive the output
    */
   static RecordParser newDelimited(String delim, Handler<Buffer> output) {
-    return RecordParserImpl.newDelimited(delim, output);
+    return RecordParserImpl.newDelimited(delim, null, output);
+  }
+
+  /**
+   * Like {@link #newDelimited(String)} but wraps the {@code stream}, the stream handlers will be set/unset
+   * when the {@link #handler(Handler)} is set.
+   * <p/>
+   * The {@code pause()}/{@code resume()} operations are propagated to the {@code stream}.
+   *
+   * @param delim  the initial delimiter string
+   * @param stream  the wrapped stream
+   */
+  static RecordParser newDelimited(String delim, ReadStream<Buffer> stream) {
+    return RecordParserImpl.newDelimited(delim, stream, null);
   }
 
   /**
@@ -80,19 +94,18 @@ static RecordParser newDelimited(String delim, Handler<Buffer> output) {
    * @param delim  the initial delimiter string
    */
   static RecordParser newDelimited(String delim) {
-    return RecordParserImpl.newDelimited(delim, null);
+    return RecordParserImpl.newDelimited(delim, null, null);
   }
 
   /**
    * Create a new {@code RecordParser} instance, initially in delimited mode, and where the delimiter can be represented
    * by the {@code Buffer} delim.
    * <p>
-   * {@code output} Will receive whole records which have been parsed.
    *
    * @param delim  the initial delimiter buffer
    */
   static RecordParser newDelimited(Buffer delim) {
-    return RecordParserImpl.newDelimited(delim, null);
+    return RecordParserImpl.newDelimited(delim,null,  null);
   }
 
   /**
@@ -103,9 +116,22 @@ static RecordParser newDelimited(Buffer delim) {
    * @param output  handler that will receive the output
    */
    static RecordParser newDelimited(Buffer delim, Handler<Buffer> output) {
-     return RecordParserImpl.newDelimited(delim, output);
+     return RecordParserImpl.newDelimited(delim, null, output);
    }
 
+  /**
+   * Like {@link #newDelimited(Buffer)} but wraps the {@code stream}, the stream handlers will be set/unset
+   * when the {@link #handler(Handler)} is set.
+   * <p/>
+   * The {@code pause()}/{@code resume()} operations are propagated to the {@code stream}.
+   *
+   * @param delim  the initial delimiter buffer
+   * @param stream  the wrapped stream
+   */
+  static RecordParser newDelimited(Buffer delim, ReadStream<Buffer> stream) {
+    return RecordParserImpl.newDelimited(delim, stream, null);
+  }
+
   /**
    * Create a new {@code RecordParser} instance, initially in fixed size mode, and where the record size is specified
    * by the {@code size} parameter.
@@ -115,7 +141,7 @@ static RecordParser newDelimited(Buffer delim, Handler<Buffer> output) {
    * @param size  the initial record size
    */
   static RecordParser newFixed(int size) {
-    return RecordParserImpl.newFixed(size, null);
+    return RecordParserImpl.newFixed(size, null, null);
   }
 
   /**
@@ -126,7 +152,20 @@ static RecordParser newFixed(int size) {
    * @param output  handler that will receive the output
    */
   static RecordParser newFixed(int size, Handler<Buffer> output) {
-    return RecordParserImpl.newFixed(size, output);
+    return RecordParserImpl.newFixed(size, null, output);
+  }
+
+  /**
+   * Like {@link #newFixed(int)} but wraps the {@code stream}, the stream handlers will be set/unset
+   * when the {@link #handler(Handler)} is set.
+   * <p/>
+   * The {@code pause()}/{@code resume()} operations are propagated to the {@code stream}.
+   *
+   * @param size  the initial record size
+   * @param stream  the wrapped stream
+   */
+  static RecordParser newFixed(int size, ReadStream<Buffer> stream) {
+    return RecordParserImpl.newFixed(size, stream, null);
   }
 
   /**
@@ -164,4 +203,19 @@ static RecordParser newFixed(int size, Handler<Buffer> output) {
    * @param buffer  a chunk of data
    */
   void handle(Buffer buffer);
+
+  @Override
+  RecordParser exceptionHandler(Handler<Throwable> handler);
+
+  @Override
+  RecordParser handler(Handler<Buffer> handler);
+
+  @Override
+  RecordParser pause();
+
+  @Override
+  RecordParser resume();
+
+  @Override
+  RecordParser endHandler(Handler<Void> endHandler);
 }
diff --git a/src/main/java/io/vertx/core/parsetools/impl/RecordParserImpl.java b/src/main/java/io/vertx/core/parsetools/impl/RecordParserImpl.java
--- a/src/main/java/io/vertx/core/parsetools/impl/RecordParserImpl.java
+++ b/src/main/java/io/vertx/core/parsetools/impl/RecordParserImpl.java
@@ -20,6 +20,7 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.Arguments;
 import io.vertx.core.parsetools.RecordParser;
+import io.vertx.core.streams.ReadStream;
 
 import java.util.Objects;
 
@@ -39,8 +40,13 @@ public class RecordParserImpl implements RecordParser {
   private byte[] delim;
   private int recordSize;
   private Handler<Buffer> output;
+  private Handler<Void> endHandler;
+  private Handler<Throwable> exceptionHandler;
 
-  private RecordParserImpl(Handler<Buffer> output) {
+  private final ReadStream<Buffer> stream;
+
+  private RecordParserImpl(ReadStream<Buffer> stream, Handler<Buffer> output) {
+    this.stream = stream;
     this.output = output;
   }
 
@@ -74,8 +80,8 @@ public static Buffer latin1StringToBytes(String str) {
    * @param delim  the initial delimiter string
    * @param output  handler that will receive the output
    */
-  public static RecordParser newDelimited(String delim, Handler<Buffer> output) {
-    return newDelimited(latin1StringToBytes(delim), output);
+  public static RecordParser newDelimited(String delim, ReadStream<Buffer> stream, Handler<Buffer> output) {
+    return newDelimited(latin1StringToBytes(delim), stream, output);
   }
 
   /**
@@ -87,8 +93,8 @@ public static RecordParser newDelimited(String delim, Handler<Buffer> output) {
    * @param delim  the initial delimiter buffer
    * @param output  handler that will receive the output
    */
-  public static RecordParser newDelimited(Buffer delim, Handler<Buffer> output) {
-    RecordParserImpl ls = new RecordParserImpl(output);
+  public static RecordParser newDelimited(Buffer delim, ReadStream<Buffer> stream, Handler<Buffer> output) {
+    RecordParserImpl ls = new RecordParserImpl(stream, output);
     ls.delimitedMode(delim);
     return ls;
   }
@@ -102,9 +108,9 @@ public static RecordParser newDelimited(Buffer delim, Handler<Buffer> output) {
    * @param size  the initial record size
    * @param output  handler that will receive the output
    */
-  public static RecordParser newFixed(int size, Handler<Buffer> output) {
+  public static RecordParser newFixed(int size, ReadStream<Buffer> stream, Handler<Buffer> output) {
     Arguments.require(size > 0, "Size must be > 0");
-    RecordParserImpl ls = new RecordParserImpl(output);
+    RecordParserImpl ls = new RecordParserImpl(stream, output);
     ls.fixedSizeMode(size);
     return ls;
   }
@@ -217,4 +223,60 @@ public void handle(Buffer buffer) {
     }
     handleParsing();
   }
+
+  private void end() {
+    Handler<Void> handler = endHandler;
+    if (handler != null) {
+      handler.handle(null);
+    }
+  }
+
+  @Override
+  public RecordParser exceptionHandler(Handler<Throwable> handler) {
+    exceptionHandler = handler;
+    return this;
+  }
+
+  @Override
+  public RecordParser handler(Handler<Buffer> handler) {
+    output = handler;
+    if (stream != null) {
+      if (handler != null) {
+        stream.endHandler(v -> end());
+        stream.exceptionHandler(err -> {
+          if (exceptionHandler != null) {
+            exceptionHandler.handle(err);
+          }
+        });
+        stream.handler(this);
+      } else {
+        stream.handler(null);
+        stream.endHandler(null);
+        stream.exceptionHandler(null);
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public RecordParser pause() {
+    if (stream != null) {
+      stream.pause();
+    }
+    return this;
+  }
+
+  @Override
+  public RecordParser resume() {
+    if (stream != null) {
+      stream.resume();
+    }
+    return this;
+  }
+
+  @Override
+  public RecordParser endHandler(Handler<Void> handler) {
+    endHandler = handler;
+    return this;
+  }
 }


Test patch:
diff --git a/src/test/java/io/vertx/test/core/RecordParserTest.java b/src/test/java/io/vertx/test/core/RecordParserTest.java
--- a/src/test/java/io/vertx/test/core/RecordParserTest.java
+++ b/src/test/java/io/vertx/test/core/RecordParserTest.java
@@ -19,14 +19,21 @@
 import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.parsetools.RecordParser;
+import io.vertx.core.streams.ReadStream;
 import org.junit.Test;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static io.vertx.test.core.TestUtils.assertNullPointerException;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author <a href="http://tfox.org">Tim Fox</a>
@@ -275,4 +282,61 @@ public void testSpreadDelimiter() {
     doTestDelimited(Buffer.buffer("start-ab-c-dddabc"), Buffer.buffer("abc"),
       new Integer[] { 18 }, Buffer.buffer("start-ab-c-ddd"));
   }
+
+  @Test
+  public void testWrapReadStream() {
+    AtomicBoolean paused = new AtomicBoolean();
+    AtomicReference<Handler<Buffer>> eventHandler = new AtomicReference<>();
+    AtomicReference<Handler<Void>> endHandler = new AtomicReference<>();
+    AtomicReference<Handler<Throwable>> exceptionHandler = new AtomicReference<>();
+    ReadStream<Buffer> original = new ReadStream<Buffer>() {
+      @Override
+      public ReadStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
+        exceptionHandler.set(handler);
+        return this;
+      }
+      @Override
+      public ReadStream<Buffer> handler(Handler<Buffer> handler) {
+        eventHandler.set(handler);
+        return this;
+      }
+      @Override
+      public ReadStream<Buffer> pause() {
+        paused.set(true);
+        return this;
+      }
+      @Override
+      public ReadStream<Buffer> resume() {
+        paused.set(false);
+        return this;
+      }
+      @Override
+      public ReadStream<Buffer> endHandler(Handler<Void> handler) {
+        endHandler.set(handler);
+        return this;
+      }
+    };
+    RecordParser parser = RecordParser.newDelimited("\r\n", original);
+    AtomicInteger ends = new AtomicInteger();
+    parser.endHandler(v -> ends.incrementAndGet());
+    List<String> records = new ArrayList<>();
+    parser.handler(record -> records.add(record.toString()));
+    assertFalse(paused.get());
+    parser.pause();
+    assertTrue(paused.get());
+    parser.resume();
+    assertFalse(paused.get());
+    eventHandler.get().handle(Buffer.buffer("first\r\nsecond\r\nthird"));
+    assertEquals(Arrays.asList("first", "second"), records);
+    assertEquals(0, ends.get());
+    Throwable cause = new Throwable();
+    exceptionHandler.get().handle(cause);
+    List<Throwable> failures = new ArrayList<>();
+    parser.exceptionHandler(failures::add);
+    exceptionHandler.get().handle(cause);
+    assertEquals(Collections.singletonList(cause), failures);
+    endHandler.get().handle(null);
+    assertEquals(Arrays.asList("first", "second"), records);
+    assertEquals(1, ends.get());
+  }
 }