[WIP] 0.6 new gekko events

core/budfox/budfox.js

@@ -30,6 +30,24 @@ var BudFox = function(config) {
 
   //    BudFox data flow:
 
+  // relay a marketUpdate event
+  this.marketDataProvider.on(
+    'marketUpdate',
+    e => this.emit('marketUpdate', e)
+  );
+
+  // relay a marketStart event
+  this.marketDataProvider.on(
+    'marketStart',
+    e => this.emit('marketStart', e)
+  );
+
+  // Output the candles
+  this.candleManager.on(
+    'candles',
+    this.pushCandles
+  );
+
   // on every `tick` retrieve trade data
   this.heart.on(
     'tick',
@@ -42,26 +60,7 @@ var BudFox = function(config) {
     this.candleManager.processTrades
   );
 
-  // Output the candles
-  this.candleManager.on(
-    'candles',
-    this.pushCandles
-  );
-
   this.heart.pump();
-
-  //    Budfox also reports:
-
-  // Trades & last trade
-  //
-  // this.marketDataProvider.on(
-  //   'trades',
-  //   this.broadcast('trades')
-  // );
-  // this.marketDataProvider.on(
-  //   'trades',
-  //   this.broadcastTrade
-  // );
 }
 
 var Readable = require('stream').Readable;
@@ -76,18 +75,4 @@ BudFox.prototype.pushCandles = function(candles) {
   _.each(candles, this.push);
 }
 
-// BudFox.prototype.broadcastTrade = function(trades) {
-//   _.defer(function() {
-//     this.emit('trade', trades.last);
-//   }.bind(this));
-// }
-
-// BudFox.prototype.broadcast = function(message) {
-//   return function(payload) {
-//     _.defer(function() {
-//       this.emit(message, payload);
-//     }.bind(this));
-//   }.bind(this);
-// }
-
 module.exports = BudFox;

core/budfox/candleCreator.js

@@ -109,7 +109,7 @@ CandleCreator.prototype.calculateCandles = function() {
 
   // catch error from high volume getTrades
   if (this.lastTrade !== undefined)
-    // create a string referencing to minute this trade happened in
+    // create a string referencing the minute this trade happened in
     var lastMinute = this.lastTrade.date.format('YYYY-MM-DD HH:mm');
 
   var candles = _.map(this.buckets, function(bucket, name) {

core/budfox/candleManager.js

@@ -21,10 +21,6 @@ var Manager = function() {
 
   this.candleCreator
     .on('candles', this.relayCandles);
-
-  this.messageFirstCandle = _.once(candle => {
-    cp.firstCandle(candle);
-  })
 };
 
 util.makeEventEmitter(Manager);
@@ -34,12 +30,6 @@ Manager.prototype.processTrades = function(tradeBatch) {
 
 Manager.prototype.relayCandles = function(candles) {
   this.emit('candles', candles);
-
-  if(!_.size(candles))
-    return;
-
-  this.messageFirstCandle(_.first(candles));
-  cp.lastCandle(_.last(candles));
 }
 
 module.exports = Manager;

core/budfox/marketDataProvider.js

@@ -33,14 +33,14 @@ Manager.prototype.retrieve = function() {
 
 
 Manager.prototype.relayTrades = function(batch) {
-  this.emit('trades', batch);
+  this.sendMarketStart(batch);
+  this.emit('marketUpdate', batch.last.date);
 
-  this.sendStartAt(batch);
-  cp.update(batch.last.date.format());
+  this.emit('trades', batch);
 }
 
-Manager.prototype.sendStartAt = _.once(function(batch) {
-  cp.startAt(batch.first.date.format())
+Manager.prototype.sendMarketStart = _.once(function(batch) {
+  this.emit('marketStart', batch.first.date);
 });
 
 module.exports = Manager;

core/candleBatcher.js

@@ -18,6 +18,7 @@ var CandleBatcher = function(candleSize) {
 
   this.candleSize = candleSize;
   this.smallCandles = [];
+  this.calculatedCandles = [];
 
   _.bindAll(this);
 }
@@ -28,22 +29,37 @@ CandleBatcher.prototype.write = function(candles) {
   if(!_.isArray(candles))
     throw 'candles is not an array';
 
+  this.emitted = 0;
+
   _.each(candles, function(candle) {
     this.smallCandles.push(candle);
     this.check();
   }, this);
+
+  return this.emitted;
 }
 
 CandleBatcher.prototype.check = function() {
   if(_.size(this.smallCandles) % this.candleSize !== 0)
     return;
 
-  this.emit('candle', this.calculate());
+  this.emitted++;
+  this.calculatedCandles.push(this.calculate());
   this.smallCandles = [];
 }
 
+CandleBatcher.prototype.flush = function() {
+  _.each(
+    this.calculatedCandles,
+    candle => this.emit('candle', candle)
+  );
+
+  this.calculatedCandles = [];
+}
+
 CandleBatcher.prototype.calculate = function() {
-  var first = this.smallCandles.shift();
+  // remove the id property of the small candle
+  var { id, ...first } = this.smallCandles.shift();
 
   first.vwp = first.vwp * first.volume;
 

core/cp.js

@@ -16,24 +16,6 @@ var message = (type, payload) => {
 }
 
 var cp = {
-  // string like: '2016-12-03T22:23:00.000Z'
-  update: latest => message('update', { latest }),
-  startAt: startAt => message('startAt', { startAt }),
-
-  // object like:
-  //
-  // {
-  //   start: '2016-12-03T22:23:00.000Z',
-  //   open: 765,
-  //   high: 765,
-  //   low: 765,
-  //   close: 765,
-  //   vwp: 765,
-  //   volume: 0,
-  //   trades: 0
-  // }
-  lastCandle: lastCandle => message('lastCandle', { lastCandle }),
-  firstCandle: firstCandle => message('firstCandle', { firstCandle }),
 
   // object like:
   //

core/emitter.js

@@ -0,0 +1,34 @@
+// Gekko uses a custom event emitter within the GekkoStream (the plugins) to guarantee
+// the correct order of events that are triggered by eachother. Turns sync events from
+// LIFO into a FIFO stack based model.
+//
+// More details here: https://forum.gekko.wizb.it/thread-56579.html
+
+const util = require('util');
+const events = require('events');
+const NativeEventEmitter = events.EventEmitter;
+
+const GekkoEventEmitter = function() {
+  NativeEventEmitter.call(this);
+  this.defferedEvents = [];
+}
+
+util.inherits(GekkoEventEmitter, NativeEventEmitter);
+
+// push to stack
+GekkoEventEmitter.prototype.deferredEmit = function(name, payload) {
+  this.defferedEvents.push({name, payload});
+}
+
+// resolve FIFO
+GekkoEventEmitter.prototype.broadcastDeferredEmit = function() {
+  if(this.defferedEvents.length === 0)
+    return false;
+
+  const event = this.defferedEvents.shift();
+
+  this.emit(event.name, event.payload);
+  return true;
+}
+
+module.exports = GekkoEventEmitter;

core/gekkoStream.js

@@ -1,30 +1,84 @@
 // Small writable stream wrapper that
 // passes data to all `candleConsumers`.
 
-var Writable = require('stream').Writable;
-var _ = require('lodash');
-var async = require('async');
+const Writable = require('stream').Writable;
+const _ = require('lodash');
+const async = require('async');
+const moment = require('moment');
 
-var util = require('./util');
-var env = util.gekkoEnv();
-var mode = util.gekkoMode();
+const util = require('./util');
+const env = util.gekkoEnv();
+const mode = util.gekkoMode();
+const config = util.getConfig();
+const log = require(util.dirs().core + 'log');
 
-var Gekko = function(candleConsumers) {
-  this.candleConsumers = candleConsumers;
+var Gekko = function(plugins) {
+  this.plugins = plugins;
+  this.candleConsumers = plugins
+    .filter(plugin => plugin.processCandle);
   Writable.call(this, {objectMode: true});
 
+  this.producers = this.plugins
+    .filter(p => p.meta.emits);
+
   this.finalize = _.bind(this.finalize, this);
 }
 
 Gekko.prototype = Object.create(Writable.prototype, {
   constructor: { value: Gekko }
 });
 
-Gekko.prototype._write = function(chunk, encoding, _done) {
-  var done = _.after(this.candleConsumers.length, _done);
-  _.each(this.candleConsumers, function(c) {
-    c.processCandle(chunk, done);
-  });
+if(config.debug) {
+  // decorate with more debug information
+  Gekko.prototype._write = function(chunk, encoding, _done) {
+
+    const start = moment();
+    var relayed = false;
+    var at = null;
+
+    const timer = setTimeout(() => {
+      if(!relayed)
+        log.error([
+          `The plugin "${at}" has not processed a candle for 1 second.`,
+          `This will cause Gekko to slow down or stop working completely.`
+        ].join(' '));
+    }, 1000);
+
+    const flushEvents = _.after(this.candleConsumers.length, () => {
+      relayed = true;
+      clearInterval(timer);
+      this.flushDefferedEvents();
+      _done();
+    });
+    _.each(this.candleConsumers, function(c) {
+      at = c.meta.name;
+      c.processCandle(chunk, flushEvents);
+    }, this);
+  }
+} else {
+  // skip decoration
+  Gekko.prototype._write = function(chunk, encoding, _done) {
+    const flushEvents = _.after(this.candleConsumers.length, () => {
+      this.flushDefferedEvents();
+      _done();
+    });
+    _.each(this.candleConsumers, function(c) {
+      c.processCandle(chunk, flushEvents);
+    }, this);
+  }
+}
+
+Gekko.prototype.flushDefferedEvents = function() {
+  const broadcasted = _.find(
+    this.producers,
+    producer => producer.broadcastDeferredEmit()
+  );
+
+  // If we braodcasted anything we might have
+  // triggered more events, recurse until we
+  // have fully broadcasted everything.
+  if(broadcasted)
+    this.flushDefferedEvents();
 }
 
 Gekko.prototype.finalize = function() {
@@ -41,12 +95,12 @@ Gekko.prototype.finalize = function() {
 
 Gekko.prototype.shutdown = function() {
   async.eachSeries(
-    this.candleConsumers,
+    this.plugins,
     function(c, callback) {
       if (c.finalize) c.finalize(callback);
       else callback();
     },
-    function() {
+    () => {
       // If we are a child process, we signal to the parent to kill the child once it is done
       // so that is has time to process all remaining events (and send report data)
       if (env === 'child-process') process.send('done');