[WIP] 0.6 new gekko events

.travis.yml

@@ -1,3 +1,3 @@
 language: node_js
 node_js:
-  - "8.0.0"
+  - "8.11.2"

appveyor.yml

@@ -2,7 +2,7 @@ init:
   - git config --global core.autocrlf input
 
 environment:
-  nodejs_version: "6"
+  nodejs_version: "9"
 
 install:
   - ps: Install-Product node $env:nodejs_version

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

@@ -10,7 +10,6 @@ var util = require(__dirname + '/../util');
 var dirs = util.dirs();
 var config = util.getConfig();
 var log = require(dirs.core + 'log');
-var cp = require(dirs.core + 'cp');
 
 var CandleCreator = require(dirs.budfox + 'candleCreator');
 
@@ -21,10 +20,6 @@ var Manager = function() {
 
   this.candleCreator
     .on('candles', this.relayCandles);
-
-  this.messageFirstCandle = _.once(candle => {
-    cp.firstCandle(candle);
-  })
 };
 
 util.makeEventEmitter(Manager);
@@ -34,12 +29,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

@@ -10,7 +10,6 @@ const util = require(__dirname + '/../util');
 
 const MarketFetcher = require('./marketFetcher');
 const dirs = util.dirs();
-const cp = require(dirs.core + 'cp');
 
 const Manager = function(config) {
 
@@ -33,14 +32,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/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;