Merge pull request #363 from uProxy/trevj-newline-fix

fix issues with newlines on SSH connections

Author: trevj

src/cloud/install/installer.ts

@@ -49,8 +49,9 @@ class CloudInstaller {
       connection.on('ready', () => {
         log.debug('logged into server');
 
-        var stdout = new queue.Queue<ArrayBuffer, void>();
-        new linefeeder.LineFeeder(stdout).setSyncHandler((line: string) => {
+        const stdoutRaw = new queue.Queue<ArrayBuffer, void>();
+        const stdout = new linefeeder.LineFeeder(stdoutRaw);
+        stdout.setSyncHandler((line: string) => {
           log.debug('STDOUT: %1', line);
           // Search for the URL anywhere in the line so we will
           // continue to work in the face of minor changes
@@ -67,8 +68,9 @@ class CloudInstaller {
           }
         });
 
-        var stderr = new queue.Queue<ArrayBuffer, void>();
-        new linefeeder.LineFeeder(stderr).setSyncHandler((line: string) => {
+        const stderrRaw = new queue.Queue<ArrayBuffer, void>();
+        const stderr = new linefeeder.LineFeeder(stderrRaw);
+        stderr.setSyncHandler((line: string) => {
           log.error('STDERR: %1', line);
         });
 
@@ -81,14 +83,16 @@ class CloudInstaller {
             return;
           }
           stream.on('end', () => {
+            stdout.flush();
+            stderr.flush();
             connection.end();
             R({
               message: 'invitation URL not found'
             });
           }).on('data', (data:Buffer) => {
-            stdout.handle(arraybuffers.bufferToArrayBuffer(data));
+            stdoutRaw.handle(arraybuffers.bufferToArrayBuffer(data));
           }).stderr.on('data', (data: Buffer) => {
-            stderr.handle(arraybuffers.bufferToArrayBuffer(data));
+            stderrRaw.handle(arraybuffers.bufferToArrayBuffer(data));
           });
         });
       }).on('error', (e: Error) => {

src/cloud/social/provider.ts

@@ -564,19 +564,21 @@ class Connection {
           return;
         }
 
-        var stdout = new queue.Queue<ArrayBuffer, void>();
-        new linefeeder.LineFeeder(stdout).setSyncHandler((line: string) => {
+        const stdoutRaw = new queue.Queue<ArrayBuffer, void>();
+        const stdout = new linefeeder.LineFeeder(stdoutRaw);
+        stdout.setSyncHandler((line: string) => {
           F(line);
         });
 
         stream.on('data', (data: Buffer) => {
-          stdout.handle(arraybuffers.bufferToArrayBuffer(data));
+          stdoutRaw.handle(arraybuffers.bufferToArrayBuffer(data));
         }).stderr.on('data', (data: Buffer) => {
           R({
             message: 'output received on STDERR: ' + data.toString()
           });
         }).on('end', () => {
           log.debug('%1: exec stream end', this.name_);
+          stdout.flush();
         });
       });
     });

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);
+  }
 }

src/zork/freedom-module.ts

@@ -148,6 +148,7 @@ function serveConnection(connection: tcp.Connection): void {
   connection.onceClosed.then((kind:tcp.SocketCloseKind) => {
     log.info('%1: closed (%2)',
       connection.connectionId, tcp.SocketCloseKind[kind]);
+    lineFeeder.flush();
   });
 }