add a flush method, for case when input terminates without a newline

Author: trevj

src/net/linefeeder.spec.ts

@@ -5,37 +5,40 @@ import linefeeder = require('./linefeeder');
 import queue = require('../handler/queue');
 
 describe('LineFeeder', function() {
-  var bufferQueue: queue.Queue<ArrayBuffer, void>;
-  var lines: linefeeder.LineFeeder;
+  let bufferQueue: queue.Queue<ArrayBuffer, void>;
+  let lines: linefeeder.LineFeeder;
 
   beforeEach(() => {
     bufferQueue = new queue.Queue<ArrayBuffer, void>();
     lines = new linefeeder.LineFeeder(bufferQueue);
   });
 
   it('one and done', (done) => {
-    var s = 'hello world';
-    bufferQueue.handle(arraybuffers.stringToArrayBuffer(s + '\n'));
+    const s = 'hello world';
+    bufferQueue.handle(arraybuffers.stringToArrayBuffer(s));
+    lines.flush();
 
     lines.setSyncNextHandler((result: string) => {
       expect(result).toEqual(s);
       done();
     });
   });
 
-  it('dangling lines', (done) => {
-    var s = 'hello world';
+  it('lines of multiple buffers', (done) => {
+    const s = 'hello world';
     bufferQueue.handle(arraybuffers.stringToArrayBuffer('hello '));
     bufferQueue.handle(arraybuffers.stringToArrayBuffer('world\n'));
+    lines.flush();
 
     lines.setSyncNextHandler((result: string) => {
       expect(result).toEqual(s);
       done();
     });
   });
 
-  it('multiple lines in one buffer', (done) => {
+  it('buffers of multiple lines', (done) => {
     bufferQueue.handle(arraybuffers.stringToArrayBuffer('a\nb\n'));
+    lines.flush();
 
     lines.setSyncNextHandler((result: string) => {
       expect(result).toEqual('a');

src/net/linefeeder.ts

@@ -9,19 +9,29 @@ export class LineFeeder extends queue.Queue<string, void> {
   private static DELIMITER = arraybuffers.decodeByte(
       arraybuffers.stringToArrayBuffer('\n'));
 
-  constructor(source: queue.Queue<ArrayBuffer, void>) {
+  private leftover_ = new ArrayBuffer(0);
+
+  constructor(private source_: queue.Queue<ArrayBuffer, void>) {
     super();
-    let leftover = new ArrayBuffer(0);
-    source.setSyncHandler((buffer: ArrayBuffer) => {
-      leftover = arraybuffers.concat([leftover, buffer]);
-      let i = arraybuffers.indexOf(leftover, LineFeeder.DELIMITER);
+    source_.setSyncHandler((buffer: ArrayBuffer) => {
+      this.leftover_ = arraybuffers.concat([this.leftover_, buffer]);
+      let i = arraybuffers.indexOf(this.leftover_, LineFeeder.DELIMITER);
       while (i !== -1) {
-        let parts = arraybuffers.split(leftover, i);
+        let parts = arraybuffers.split(this.leftover_, i);
         let line = arraybuffers.arrayBufferToString(parts[0]);
-        leftover = parts[1].slice(1);
-        i = arraybuffers.indexOf(leftover, LineFeeder.DELIMITER);
+        this.leftover_ = parts[1].slice(1);
+        i = arraybuffers.indexOf(this.leftover_, LineFeeder.DELIMITER);
         this.handle(line);
       }
     });
   }
+
+  // Causes any pending line to be emitted. Intended to be called when the
+  // underlying stream has terminated, possibly without any terminating
+  // newline.
+  public flush = () => {
+    var s = arraybuffers.arrayBufferToString(this.leftover_);
+    this.leftover_ = new ArrayBuffer(0);
+    this.handle(s);
+  }
 }