Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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();
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.