keepAlive.js

const setTimeout = require('timers').setTimeout;
const clearTimeout = require('timers').clearTimeout;
const WebSocket = require('ws');
const Bluebird = require('bluebird');
const Queue = require('promise-queue');

const binance = require('../node-binance-api.js');
binance.options({
  APIKEY: '<api key>',
  APISECRET: '<api secret>',
});

let watchdog = {};
const stream = 'wss://stream.binance.com:9443/ws/';

var queue = new Queue(1, Infinity);

let subscriptions = {};

function subscribe(key) {
  return Bluebird.delay(10).then(() =>
    queue.add(() => {
      const endpoint = key.toLowerCase() + '@aggTrade';
      console.log('Subscribing to endpoing: ', endpoint);
      const ws = new WebSocket(stream + endpoint);
      ws.on('open', function() {
        console.log('subscribe(' + endpoint + ')');
        subscriptions[key].isAlive = true;
        subscriptions[key].lastActivity = new Date();
        watchdog[key] = new Date().getTime();
      });
      ws.on('close', function() {
        console.log('Closed ticker!! ', key);
      });
      ws.on('error', function() {
        console.log('Error ticker!! ', key);
      });
      ws.on('message', function(data) {
        //console.log(data);
        const trades = JSON.parse(data);
        const {
          e: eventType,
          E: eventTime,
          s: symbol,
          p: price,
          q: quantity,
          m: maker,
          a: tradeId,
        } = trades;
        watchdog[symbol] = new Date().getTime();
      });
      function heartbeat() {
        subscriptions[key].isAlive = true;
        subscriptions[key].lastActivity = new Date();
      }
      ws.on('pong', heartbeat);

      function watcher() {
        const wss = subscriptions[key];
        clearTimeout(wss.heartbeat);
        wss.heartbeat = setTimeout(() => {
          if (!wss.isAlive) {
            console.log(key, ' Watchdog is worried');
            console.log('DEAD SOCKET!! TERMINATE');
            wss.terminate();
            subscribe(key);
          } else {
            wss.isAlive = false;
            subscriptions[key] = wss;
            wss.ping(() => {});
            watcher();
          }
        }, 20000);
      }

      subscriptions[key] = ws;
      watcher();

      return ws;
    }),
  );
}

// Trade history
binance.prices((error, prices) => {
  console.log('Prices =>', prices);
  const keys = Object.keys(prices).filter(key => key.indexOf('BTC') >= 0);
  Bluebird.each(keys, key => (subscriptions[key] = subscribe(key)));
});

// This piece is just here to show it's working and staying alive
function watcher() {
  setTimeout(() => {
    const keys = Object.keys(subscriptions);
    console.log('RUNN SCAN ============================================');
    keys.forEach(key => {
      const ws = subscriptions[key];
      console.log(
        'WS: ',
        key,
        'IS ACTIVE: ',
        ws.isAlive,
        ' Last Activity: ',
        ws.lastActivity,
      );
    });

    watcher();
  }, 1000 * 60);
}

watcher();