-
Notifications
You must be signed in to change notification settings - Fork 0
/
ApiStream.js
107 lines (93 loc) · 3.07 KB
/
ApiStream.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
var WebSocket = require('ws');
function ApiStream(opts) {
var wss;
var http_server;
var self = this;
function debug(msg) {
if(process.stdout.clearLine) {
process.stdout.clearLine();
console.log('\r' + msg);
} else {
console.log(msg);
}
}
this.send_all = function(data) {
wss.clients.forEach(function(ws) {
if(ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify(data));
}
});
}
this.start = function(app) {
var ws_server_config = {};
if(app) {
http_server = require('http').createServer(app);
ws_server_config.server = http_server;
} else {
ws_server_config.port = opts.server.ws_port;
}
ws_server_config.path = opts.server.ws_path || '/api';
if(opts.server.ws_deflate) {
ws_server_config.perMessageDeflate = {
zlibDeflateOptions: {
chunkSize: 1024,
memLevel: 7,
level: 3
},
zlibInflateOptions: {
chunkSize: 10 * 1024
},
clientNoContextTakeover: true,
serverNoContextTakeover: true,
serverMaxWindowBits: 10,
concurrencyLimit: 10,
threshold: 1024
};
}
wss = new WebSocket.Server(ws_server_config);
function get_client_count() {
var client_count = 0;
wss.clients.forEach(function(ws) {
client_count++;
});
return client_count;
}
function heartbeat() {
this.isAlive = true;
}
function noop() {}
wss.on('connection', function connection(ws, req) {
ws.subscribe = {};
ws.isAlive = true;
ws.ip = (req.headers['x-forwarded-for'] || '').split(/\s*,\s*/)[0] || req.connection.remoteAddress;
debug('ws connect ip=' + ws.ip + ' count=' + get_client_count());
ws.on('pong', heartbeat);
ws.on('close', function() {
debug('ws close ip=' + ws.ip + ' count=' + get_client_count());
});
ws.on('message', function(message) {
debug(message);
try {
var data = JSON.parse(message);
} catch(e) {
debug(e.name + ': ' + e.message);
}
});
});
if(opts.server.ws_heartbeat == null || opts.server.ws_heartbeat != 0) {
var heartbeat_intval = setInterval(function() {
wss.clients.forEach(function(ws) {
if(ws.isAlive === false) {
return ws.terminate();
}
ws.isAlive = false;
ws.ping(noop);
});
}, 10000);
}
if(app) {
http_server.listen(opts.server.http_port);
}
}
}
module.exports = ApiStream;