Allow catching navdata errors

lib/Client.js

@@ -1,15 +1,22 @@
-Client.UdpControl = require('./control/UdpControl');
-Client.Repl       = require('./Repl');
+var EventEmitter = require('events').EventEmitter;
+var util         = require('util');
+
+Client.UdpControl       = require('./control/UdpControl');
+Client.Repl             = require('./Repl');
+Client.UdpNavdataStream = require('./navdata/UdpNavdataStream');
 
 module.exports = Client;
+util.inherits(Client, EventEmitter);
 function Client(options) {
+  EventEmitter.call(this);
+
   options = options || {};
 
   this._udpControl = options.udpControl || new Client.UdpControl(options);
+  this._udpNavdatasStream = options.udpNavdataStream || new Client.UdpNavdataStream(options);
   this._interval   = null;
   this._ref        = {};
   this._pcmd       = {};
-  this._configs    = {};
   this._repeaters  = [];
 }
 
@@ -21,6 +28,10 @@ Client.prototype.createRepl = function() {
 
 Client.prototype.resume = function() {
   this._setInterval(30);
+
+  this._udpNavdatasStream.removeAllListeners();
+  this._udpNavdatasStream.resume();
+  this._udpNavdatasStream.on('data', this.emit.bind(this, 'navdata'));
 };
 
 Client.prototype._setInterval = function(duration) {

lib/navdata/UdpNavdataStream.js

@@ -11,14 +11,15 @@ function UdpNavdataStream(options) {
 
   options = options || {};
 
-  this.readable      = true;
-  this._socket       = options.socket || dgram.createSocket('udp4');
-  this._port         = options.port || constants.ports.NAVDATA;
-  this._ip           = options.ip || constants.DEFAULT_DRONE_IP;
-  this._initialized  = false;
-  this._parseNavdata = options.parser || parseNavdata;
-  this._timeout      = options.timeout || 100;
-  this._timer        = undefined;
+  this.readable        = true;
+  this._socket         = options.socket || dgram.createSocket('udp4');
+  this._port           = options.port || constants.ports.NAVDATA;
+  this._ip             = options.ip || constants.DEFAULT_DRONE_IP;
+  this._initialized    = false;
+  this._parseNavdata   = options.parser || parseNavdata;
+  this._timeout        = options.timeout || 100;
+  this._timer          = undefined;
+  this._sequenceNumber = 0;
 }
 
 UdpNavdataStream.prototype.resume = function() {
@@ -54,6 +55,21 @@ UdpNavdataStream.prototype._setTimeout = function() {
 };
 
 UdpNavdataStream.prototype._handleMessage = function(buffer) {
-  this.emit('data', this._parseNavdata(buffer));
+  try {
+    var navdata = this._parseNavdata(buffer);
+  } catch (err) {
+    // avoid 'error' causing an exception when nobody is listening
+    if (this.listeners('error').length > 0) {
+      this.emit('error', err);
+    }
+    return;
+  }
+
+  // Ignore out of order messages
+  if (navdata.sequenceNumber > this._sequenceNumber) {
+    this._sequenceNumber = navdata.sequenceNumber;
+    this.emit('data', navdata);
+  }
+
   this._setTimeout();
 };

test/unit/navdata/test-UdpNavdataStream.js

@@ -85,6 +85,7 @@ test('UdpNavdataStream', {
 
     this.clock.tick(this.fakeTimeout - 1);
 
+    this.fakeParser.returns({});
     this.fakeSocket.emit('message', new Buffer(0));
 
     this.clock.tick(1);
@@ -94,10 +95,9 @@ test('UdpNavdataStream', {
     assert.equal(this.fakeSocket.send.callCount, 2);
   },
 
-
   'incoming messages are parsed': function() {
     var fakeBuffer  = new Buffer([1, 2, 3]);
-    var fakeNavdata = {fake: 'navdata'};
+    var fakeNavdata = {fake: 'navdata', sequenceNumber: 1};
     var dataSpy     = sinon.spy();
 
     this.fakeParser.returns(fakeNavdata);
@@ -114,6 +114,62 @@ test('UdpNavdataStream', {
     assert.strictEqual(dataSpy.getCall(0).args[0], fakeNavdata);
   },
 
+  'old navdata messages are ignored': function() {
+    var fakeNavdataA = {sequenceNumber: 1};
+    var fakeNavdataB = {sequenceNumber: 2};
+    var fakeNavdataC = {sequenceNumber: 3};
+
+    this.fakeParser.withArgs(1).returns(fakeNavdataA);
+    this.fakeParser.withArgs(2).returns(fakeNavdataB);
+    this.fakeParser.withArgs(3).returns(fakeNavdataC);
+
+    var dataSpy = sinon.spy();
+    this.stream.on('data', dataSpy);
+
+    this.stream.resume();
+    this.fakeSocket.emit('message', 1);
+    this.fakeSocket.emit('message', 3);
+    this.fakeSocket.emit('message', 2);
+
+    assert.equal(this.fakeParser.callCount, 3);
+    assert.equal(dataSpy.callCount, 2);
+
+    assert.equal(dataSpy.getCall(0).args[0].sequenceNumber, 1);
+    assert.equal(dataSpy.getCall(1).args[0].sequenceNumber, 3);
+  },
+
+  'navdata errors are ignored by default': function() {
+    this.fakeParser.throws(new Error('bad'));
+
+    var dataSpy = sinon.spy();
+    this.stream.on('data', dataSpy);
+
+    this.stream.resume();
+    this.fakeSocket.emit('message', 1);
+
+    assert.equal(this.fakeParser.callCount, 1);
+    assert.equal(dataSpy.callCount, 0);
+  },
+
+  'navdata errors are emitted if there is an error handler': function() {
+    var fakeErr = new Error('bad');
+    this.fakeParser.throws(fakeErr);
+
+    var dataSpy = sinon.spy();
+    var errorSpy = sinon.spy();
+
+    this.stream.on('data', dataSpy);
+    this.stream.on('error', errorSpy);
+
+    this.stream.resume();
+    this.fakeSocket.emit('message', 1);
+
+    assert.equal(this.fakeParser.callCount, 1);
+    assert.equal(dataSpy.callCount, 0);
+    assert.equal(errorSpy.callCount, 1);
+    assert.strictEqual(errorSpy.getCall(0).args[0], fakeErr);
+  },
+
   'destroy() cleans up': function() {
     this.stream.destroy();
     assert.equal(this.fakeSocket.close.callCount, 1);

test/unit/test-Client.js

@@ -1,8 +1,9 @@
-var common = require('../common');
-var assert = require('assert');
-var test   = require('utest');
-var sinon  = require('sinon');
-var Client = require(common.lib + '/Client');
+var common       = require('../common');
+var assert       = require('assert');
+var test         = require('utest');
+var sinon        = require('sinon');
+var Client       = require(common.lib + '/Client');
+var EventEmitter = require('events').EventEmitter;
 
 test('Client', {
   before: function() {
@@ -13,8 +14,12 @@ test('Client', {
     this.fakeUdpControl.animate     = sinon.stub();
     this.fakeUdpControl.flush       = sinon.stub();
 
+    this.fakeUdpNavdataStream        = new EventEmitter;
+    this.fakeUdpNavdataStream.resume = sinon.stub();
+
     this.client = new Client({
-      udpControl: this.fakeUdpControl,
+      udpControl       : this.fakeUdpControl,
+      udpNavdataStream : this.fakeUdpNavdataStream,
     });
 
     this.clock = sinon.useFakeTimers();
@@ -24,14 +29,45 @@ test('Client', {
     this.clock.restore();
   },
 
-  'resume calls _setInterval': function() {
+  'resume() calls _setInterval': function() {
     sinon.spy(this.client, '_setInterval');
     this.client.resume();
 
     assert.equal(this.client._setInterval.callCount, 1);
     assert.equal(this.client._setInterval.getCall(0).args[0], 30);
   },
 
+  'navdata events are proxied': function() {
+    var fakeNavdata = {fake: 'navdata'};
+    this.client.resume();
+
+    assert.equal(this.fakeUdpNavdataStream.resume.callCount, 1);
+
+    var gotNavdata;
+    this.client.on('navdata', function(navdata) {
+      gotNavdata = navdata;
+    });
+
+    this.fakeUdpNavdataStream.emit('data', fakeNavdata);
+
+    assert.strictEqual(gotNavdata, fakeNavdata);
+  },
+
+  'resume() is idempotent': function() {
+    var fakeNavdata = {fake: 'navdata'};
+    this.client.resume();
+    this.client.resume();
+
+    var eventCount = 0;
+    this.client.on('navdata', function(navdata) {
+      eventCount++;
+    });
+
+    this.fakeUdpNavdataStream.emit('data', fakeNavdata);
+
+    assert.strictEqual(eventCount, 1);
+  },
+
   'options are passed to internal UdpControl': function() {
     var options = {fake: 'options'};