StreamParserObject::eof returns an Option

.haxerc

@@ -1,4 +1,4 @@
 {
-  "version": "4.2.1",
+  "version": "4.2.2",
   "resolveLibs": "scoped"
 }

haxe_libraries/tink_chunk.hxml

@@ -1,3 +1,3 @@
-# @install: lix --silent download "gh://github.com/haxetink/tink_chunk#d58e35ce6985a9e40e76a9771e8eee30c6efa5aa" into tink_chunk/0.4.0/github/d58e35ce6985a9e40e76a9771e8eee30c6efa5aa
--cp ${HAXE_LIBCACHE}/tink_chunk/0.4.0/github/d58e35ce6985a9e40e76a9771e8eee30c6efa5aa/src
+# @install: lix --silent download "gh://github.com/haxetink/tink_chunk#66ced3303a377829362352dd3f60bc2ff836d6a9" into tink_chunk/0.4.0/github/66ced3303a377829362352dd3f60bc2ff836d6a9
+-cp ${HAXE_LIBCACHE}/tink_chunk/0.4.0/github/66ced3303a377829362352dd3f60bc2ff836d6a9/src
 -D tink_chunk=0.4.0

src/tink/io/Source.hx

@@ -197,21 +197,16 @@ class RealSourceTools {
   static public function parse<R>(s:RealSource, p:StreamParser<R>):Promise<Pair<R, RealSource>>
     return StreamParser.parse(s, p).map(function (r) return switch r {
       case Parsed(data, rest): Success(new Pair(data, rest));
+      case Finished: Failure(new Error('No result'));
       case Invalid(e, _) | Broke(e): Failure(e);
     });
 
   static public function split(src:RealSource, delim:Chunk):SplitResult<Error> {
     var s = parse(src, new Splitter(delim));
     // TODO: make all these lazy
     return {
-      before: Stream #if cs .dirtyPromise #else .promise #end(s.next(function(p):SourceObject<Error> return switch p.a {
-        case Some(chunk): (chunk:RealSource);
-        case None: src;
-      })),
-      delimiter: s.next(function(p) return switch p.a {
-        case Some(_): delim;
-        case None: new Error(NotFound, 'Delimiter not found');
-      }),
+      before: Stream #if cs .dirtyPromise #else .promise #end(s.next(function(p):SourceObject<Error> return (p.a:RealSource))),
+      delimiter: s.swap(delim),
       after: Stream #if cs .dirtyPromise #else .promise #end(s.next(function(p):SourceObject<Error> return p.b)),
     }
   }
@@ -239,6 +234,7 @@ class IdealSourceTools {
   static public function parse<R>(s:IdealSource, p:StreamParser<R>):Promise<Pair<R, IdealSource>>
     return StreamParser.parse(s, p).map(function (r) return switch r {
       case Parsed(data, rest): Success(new Pair(data, rest));
+      case Finished: Failure(new Error('No result'));
       case Invalid(e, _): Failure(e);
     });
 

src/tink/io/StreamParser.hx

@@ -15,6 +15,7 @@ enum ParseStep<Result> {
 }
 
 enum ParseResult<Result, Quality> {
+  Finished:ParseResult<Result, Quality>;
   Parsed(data:Result, rest:Source<Quality>):ParseResult<Result, Quality>;
   Invalid(e:Error, rest:Source<Quality>):ParseResult<Result, Quality>;
   Broke(e:Error):ParseResult<Result, Error>;
@@ -70,14 +71,16 @@ abstract StreamParser<Result>(StreamParserObject<Result>) from StreamParserObjec
         Future.sync(Invalid(e, mk(rest)));
       case Failed(e):
         Future.sync(Broke(e));
-      case Depleted if(cursor.currentPos < cursor.length): 
-        Future.sync(Parsed(finish(), mk(Chunk.EMPTY)));
+      // case Depleted if(cursor.currentPos < cursor.length): 
+      //   Future.sync(Parsed(finish(), mk(Chunk.EMPTY)));
       case Depleted if(!resume):
         Future.sync(Parsed(finish(), flush()));
       case Depleted:
         switch p.eof(cursor) {
-          case Success(result):
+          case Success(Some(result)):
             consume(result).map(function (_) return Parsed(finish(), flush()));
+          case Success(None):
+            Future.sync(Finished);
           case Failure(e):
             Future.sync(Invalid(e, flush()));
         }     
@@ -98,6 +101,8 @@ abstract StreamParser<Result>(StreamParserObject<Result>) from StreamParserObjec
         step(End);
       else 
         parse(s, p).handle(function(o) switch o {
+          case Finished:
+            step(End);
           case Parsed(result, rest):
             s = rest;
             step(Link(result, Generator.stream(next)));
@@ -107,59 +112,59 @@ abstract StreamParser<Result>(StreamParserObject<Result>) from StreamParserObjec
   }
 }
 
-class Splitter extends BytewiseParser<Option<Chunk>> {
+class Splitter implements StreamParserObject<Chunk> {
   var delim:Chunk;
-  var buf = Chunk.EMPTY;
+  var scanned:Chunk = Chunk.EMPTY;
+
   public function new(delim) {
     this.delim = delim;
   }
-  override function read(char:Int):ParseStep<Option<Chunk>> {
-
-    if(char == -1) return Done(None);
-
-    buf = buf & String.fromCharCode(char);
-    return if(buf.length >= delim.length) {
-      var bcursor = buf.cursor();
-      bcursor.moveBy(buf.length - delim.length);
-      var dcursor = delim.cursor();
-
-      for(i in 0...delim.length) {
-        if(bcursor.currentByte != dcursor.currentByte) {
-          return Progressed;
-        }
-        else {
-          bcursor.next();
-          dcursor.next();
-        }
-      }
-      var out = Done(Some(buf.slice(0, bcursor.currentPos - delim.length)));
-      buf = Chunk.EMPTY;
-      return out;
-
-    } else {
-
-      Progressed;
-
+
+  public function progress(cursor:ChunkCursor) {
+    return switch cursor.seek(delim) {
+      case Some(chunk):
+        final result = scanned & chunk;
+        scanned = Chunk.EMPTY;
+        Done(result);
+      case None:
+        // move cursor to end, but leave (delim.length-1) bytes because the perfect match could happen when combined with the very first byte of the next chunk
+        scanned &= cursor.sweepTo(cursor.length - delim.length + 1);
+        Progressed;
     }
   }
+
+  public function eof(rest:ChunkCursor) {
+    return Success(rest.length == 0 ? None : Some(rest.seek(delim).or(() -> scanned & rest.right())));
+  }
 }
 
 class SimpleBytewiseParser<Result> extends BytewiseParser<Result> {
 
   var _read:Int->ParseStep<Result>;
+  var _readEof:Void->Outcome<Option<Result>, Error>;
 
-  public function new(f)
-    this._read = f;
+  public function new(read, readEof) {
+    this._read = read;
+    this._readEof = readEof;
+
+  }
 
   override public function read(char:Int)
     return _read(char); 
+
+  override function readEof():Outcome<Option<Result>, Error>
+    return _readEof();
 }
 
 class BytewiseParser<Result> implements StreamParserObject<Result> { 
 
   function read(char:Int):ParseStep<Result> {
     return throw 'abstract';
   }
+
+  function readEof():Outcome<Option<Result>, Error> {
+    return throw 'abstract';
+  }
 
   public function progress(cursor:ChunkCursor) {
 
@@ -176,16 +181,10 @@ class BytewiseParser<Result> implements StreamParserObject<Result> {
   }
 
   public function eof(rest:ChunkCursor) 
-    return switch read( -1) {
-      case Progressed: Failure(new Error(UnprocessableEntity, 'Unexpected end of input'));
-      case Done(r): Success(r);
-      case Failed(e): Failure(e);
-    }
-
-
+    return readEof();
 }
 
 interface StreamParserObject<Result> {
   function progress(cursor:ChunkCursor):ParseStep<Result>;
-  function eof(rest:ChunkCursor):Outcome<Result, Error>;
+  function eof(rest:ChunkCursor):Outcome<Option<Result>, Error>;
 }

tests/ParserTest.hx

@@ -13,7 +13,7 @@ class ParserTest {
 	@:describe('Should halt properly after consuming a result (issue #23)')
 	public function properHalt() {
 		var src:IdealSource = '1';
-		return src.parse(new SimpleBytewiseParser(function(c) return if(c == -1) Progressed else Done(c)))
+		return src.parse(new SimpleBytewiseParser(c -> if(c == -1) Progressed else Done(c), () -> Success(None)))
 			.next(function(o) return assert(o.a == '1'.code));
 	}
 }

tests/SourceTest.hx

@@ -95,7 +95,7 @@ class SourceTest {
 		var split = s1.split('3');
 		split.before.all().handle(function(chunk) asserts.assert(chunk == '12'));
 		split.after.all().handle(function(chunk) asserts.assert(chunk == ''));
-		split.delimiter.handle(function(o) asserts.assert(o.orUse(Chunk.EMPTY) == ''));
+		split.delimiter.handle(function(o) asserts.assert(o.orUse(Chunk.EMPTY) == '3'));
 
 		var s1:IdealSource = '12131415';
 		var split = s1.split('1');
@@ -116,7 +116,7 @@ class SourceTest {
 
 	public function parseStream() {
 		var s1:IdealSource = '01234';
-		var stream = s1.parseStream(new SimpleBytewiseParser(function(c) return Done(c)));
+		var stream = s1.parseStream(new SimpleBytewiseParser(c -> Done(c), () -> Success(None)));
 		stream.collect().handle(function(o) switch o {
 			case Success(items):
 				asserts.assert(items.length == 5);
@@ -126,7 +126,7 @@ class SourceTest {
 
 		var s1:IdealSource = '012';
 		var s2:IdealSource = '34';
-		var stream = s1.append(s2).parseStream(new SimpleBytewiseParser(function(c) return Done(c)));
+		var stream = s1.append(s2).parseStream(new SimpleBytewiseParser(c -> Done(c), () -> Success(None)));
 		stream.collect().handle(function(o) switch o {
 			case Success(items):
 				asserts.assert(items.length == 5);