Merge pull request #332 from uProxy/trevj-cloud-fixes

better error handling and logging for cloud

Author: trevj

Gruntfile.coffee

@@ -125,6 +125,7 @@ taskManager.add 'browserifySpecs', [
   'browserify:datachannelSpec'
   'browserify:poolSpec'
   'browserify:tcpSpec'
+  'browserify:linefeederSpec'
   'browserify:queueSpec'
   'browserify:turnFrontEndMessagesSpec'
   'browserify:turnFrontEndSpec'
@@ -589,6 +590,7 @@ module.exports = (grunt) ->
       churnCovSpec: Rule.addCoverageToBrowserify(Rule.browserifySpec 'churn/churn')
       handlerSpec: Rule.browserifySpec 'handler/queue'
       handlerCovSpec: Rule.addCoverageToBrowserify(Rule.browserifySpec 'handler/queue')
+      linefeederSpec: Rule.browserifySpec 'net/linefeeder'
       loggingProviderSpec: Rule.browserifySpec 'loggingprovider/loggingprovider'
       loggingProviderCovSpec: Rule.addCoverageToBrowserify(Rule.browserifySpec 'loggingprovider/loggingprovider')
       loggingSpec: Rule.browserifySpec 'logging/logging'

src/cloud/social/provider.ts

@@ -4,7 +4,9 @@
 /// <reference path='../../../../third_party/typings/ssh2/ssh2.d.ts' />
 
 import arraybuffers = require('../../arraybuffers/arraybuffers');
+import linefeeder = require('../../net/linefeeder');
 import logging = require('../../logging/logging');
+import queue = require('../../handler/queue');
 
 // https://github.com/borisyankov/DefinitelyTyped/blob/master/ssh2/ssh2-tests.ts
 import * as ssh2 from 'ssh2';
@@ -280,7 +282,7 @@ export class CloudSocialProvider {
           //       the instance (safe because all we've done is run ping).
           log.info('new proxying session %1', payload.proxyingId);
           if (!(destinationClientId in this.savedContacts_)) {
-            return Promise.reject('unknown client ' + destinationClientId);
+            return Promise.reject(new Error('unknown client ' + destinationClientId));
           }
           return this.reconnect_(this.savedContacts_[destinationClientId].invite).then(
               (connection: Connection) => {
@@ -293,14 +295,14 @@ export class CloudSocialProvider {
               connection.sendMessage(JSON.stringify(payload));
             });
           } else {
-            return Promise.reject('unknown client ' + destinationClientId);
+            return Promise.reject(new Error('unknown client ' + destinationClientId));
           }
         }
       } else {
-        return Promise.reject('message has no or wrong type field');
+        return Promise.reject(new Error('message has no or wrong type field'));
       }
     } catch (e) {
-      return Promise.reject('could not de-serialise message: ' + e.message);
+      return Promise.reject(new Error('could not de-serialise message: ' + e.message));
     }
   }
 
@@ -349,7 +351,7 @@ export class CloudSocialProvider {
         // Return nothing for type checking purposes.
       });
     } catch (e) {
-      return Promise.reject('could not parse invite code: ' + e.message);
+      return Promise.reject(new Error('could not parse invite code: ' + e.message));
     }
   }
 
@@ -395,7 +397,7 @@ class Connection {
   // TODO: timeout
   public connect = (): Promise<void> => {
     if (this.state_ !== ConnectionState.NEW) {
-      return Promise.reject('can only connect in NEW state');
+      return Promise.reject(new Error('can only connect in NEW state'));
     }
     this.state_ = ConnectionState.CONNECTING;
 
@@ -424,49 +426,45 @@ class Connection {
           ZORK_HOST, ZORK_PORT, (e: Error, stream: ssh2.Channel) => {
             if (e) {
               this.close();
-              R('error establishing tunnel: ' + e.toString());
+              R(new Error('error establishing tunnel: ' + e.toString()));
+              return;
             }
             this.setState_(ConnectionState.WAITING_FOR_PING);
 
             this.stream_ = stream;
 
-            // TODO: add error handler for stream
-            let leftover = new ArrayBuffer(0);
-            this.stream_.on('data', (data: Buffer) => {
-              leftover = arraybuffers.concat([leftover,
-                  arraybuffers.bufferToArrayBuffer(data)]);
-              let i = arraybuffers.indexOf(leftover, Connection.COMMAND_DELIMITER);
-              while (i !== -1) {
-                let parts = arraybuffers.split(leftover, i);
-                let reply = arraybuffers.arrayBufferToString(parts[0]).trim();
-                leftover = parts[1].slice(1);
-                i = arraybuffers.indexOf(leftover, Connection.COMMAND_DELIMITER);
-
-                switch (this.state_) {
-                  case ConnectionState.WAITING_FOR_PING:
-                    if (reply === 'ping') {
-                      this.setState_(ConnectionState.ESTABLISHED);
-                      F();
-                    } else {
-                      this.close();
-                      R(new Error('did not receive ping from server on login: ' +
-                          reply));
-                    }
-                    break;
-                  case ConnectionState.ESTABLISHED:
-                    try {
-                      this.received_(JSON.parse(reply));
-                    } catch (e) {
-                      log.warn('%1: could not de-serialise signalling message: %2',
-                          this.invite_, reply);
-                    }
-                    break;
-                  default:
-                    log.warn('%1: did not expect message in state %2: %3',
-                        this.name_, ConnectionState[this.state_], reply);
+            var bufferQueue = new queue.Queue<ArrayBuffer, void>();
+            new linefeeder.LineFeeder(bufferQueue).setSyncHandler((reply: string) => {
+              log.debug('%1: received message: %2', this.name_, reply);
+              switch (this.state_) {
+                case ConnectionState.WAITING_FOR_PING:
+                  if (reply === 'ping') {
+                    this.setState_(ConnectionState.ESTABLISHED);
+                    F();
+                  } else {
                     this.close();
-                }
+                    R(new Error('did not receive ping from server on login: ' +
+                      reply));
+                  }
+                  break;
+                case ConnectionState.ESTABLISHED:
+                  try {
+                    this.received_(JSON.parse(reply));
+                  } catch (e) {
+                    log.warn('%1: could not de-serialise signalling message: %2',
+                      this.invite_, reply);
+                  }
+                  break;
+                default:
+                  log.warn('%1: did not expect message in state %2: %3',
+                    this.name_, ConnectionState[this.state_], reply);
+                  this.close();
               }
+            });
+
+            // TODO: add error handler for stream
+            stream.on('data', (buffer: Buffer) => {
+              bufferQueue.handle(arraybuffers.bufferToArrayBuffer(buffer));
             }).on('error', (e: Error) => {
               // This occurs when:
               //  - host cannot be reached, e.g. non-existant hostname
@@ -529,7 +527,7 @@ class Connection {
   // Fetches the server's description, i.e. /banner.
   public getBanner = (): Promise<string> => {
     if (this.state_ !== ConnectionState.ESTABLISHED) {
-      return Promise.reject('can only fetch banner in ESTABLISHED state');
+      return Promise.reject(new Error('can only fetch banner in ESTABLISHED state'));
     }
     return new Promise<string>((F, R) => {
       this.client_.exec('cat /banner', (e: Error, stream: ssh2.Channel) => {

src/net/linefeeder.spec.ts

@@ -0,0 +1,48 @@
+/// <reference path='../../../third_party/typings/jasmine/jasmine.d.ts' />
+
+import arraybuffers = require('../arraybuffers/arraybuffers');
+import linefeeder = require('./linefeeder');
+import queue = require('../handler/queue');
+
+describe('LineFeeder', function() {
+  var bufferQueue: queue.Queue<ArrayBuffer, void>;
+  var 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'));
+
+    lines.setSyncNextHandler((result: string) => {
+      expect(result).toEqual(s);
+      done();
+    });
+  });
+
+  it('dangling lines', (done) => {
+    var s = 'hello world';
+    bufferQueue.handle(arraybuffers.stringToArrayBuffer('hello '));
+    bufferQueue.handle(arraybuffers.stringToArrayBuffer('world\n'));
+
+    lines.setSyncNextHandler((result: string) => {
+      expect(result).toEqual(s);
+      done();
+    });
+  });
+
+  it('multiple lines in one buffer', (done) => {
+    bufferQueue.handle(arraybuffers.stringToArrayBuffer('a\nb\n'));
+
+    lines.setSyncNextHandler((result: string) => {
+      expect(result).toEqual('a');
+      lines.setSyncNextHandler((result: string) => {
+        expect(result).toEqual('b');
+        done();
+      });
+    });
+  });
+});

src/net/linefeeder.ts

@@ -0,0 +1,27 @@
+/// <reference path='../../../third_party/typings/es6-promise/es6-promise.d.ts' />
+
+import arraybuffers = require('../arraybuffers/arraybuffers');
+import queue = require('../handler/queue');
+
+// Transforms a raw network-like ArrayBuffer-based queue into
+// a telnet-style string-based queue.
+export class LineFeeder extends queue.Queue<string, void> {
+  private static DELIMITER = arraybuffers.decodeByte(
+      arraybuffers.stringToArrayBuffer('\n'));
+
+  constructor(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);
+      while (i !== -1) {
+        let parts = arraybuffers.split(leftover, i);
+        let line = arraybuffers.arrayBufferToString(parts[0]);
+        leftover = parts[1].slice(1);
+        i = arraybuffers.indexOf(leftover, LineFeeder.DELIMITER);
+        this.handle(line);
+      }
+    });
+  }
+}