Permalink
Browse files

lib: refactor internal/socket_list

PR-URL: #11406
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Michaël Zasso <targos@protonmail.com>
  • Loading branch information...
jasnell committed Feb 15, 2017
1 parent f04387e commit 88c3f57cc35f035f793e4b5dad6e8e880eb6003a
Showing with 80 additions and 79 deletions.
  1. +80 −79 lib/internal/socket_list.js
View
@@ -1,104 +1,105 @@
'use strict';
module.exports = {SocketListSend, SocketListReceive};
const EventEmitter = require('events');
const util = require('util');
// This object keep track of the socket there are sended
function SocketListSend(slave, key) {
EventEmitter.call(this);
class SocketListSend extends EventEmitter {
constructor(child, key) {
super();
this.key = key;
this.child = child;
}
this.key = key;
this.slave = slave;
}
util.inherits(SocketListSend, EventEmitter);
_request(msg, cmd, callback) {
var self = this;
if (!this.child.connected) return onclose();
this.child.send(msg);
SocketListSend.prototype._request = function(msg, cmd, callback) {
var self = this;
function onclose() {
self.child.removeListener('internalMessage', onreply);
callback(new Error('child closed before reply'));
}
if (!this.slave.connected) return onclose();
this.slave.send(msg);
function onreply(msg) {
if (!(msg.cmd === cmd && msg.key === self.key)) return;
self.child.removeListener('disconnect', onclose);
self.child.removeListener('internalMessage', onreply);
function onclose() {
self.slave.removeListener('internalMessage', onreply);
callback(new Error('Slave closed before reply'));
callback(null, msg);
}
this.child.once('disconnect', onclose);
this.child.on('internalMessage', onreply);
}
function onreply(msg) {
if (!(msg.cmd === cmd && msg.key === self.key)) return;
self.slave.removeListener('disconnect', onclose);
self.slave.removeListener('internalMessage', onreply);
close(callback) {
this._request({
cmd: 'NODE_SOCKET_NOTIFY_CLOSE',
key: this.key
}, 'NODE_SOCKET_ALL_CLOSED', callback);
}
callback(null, msg);
getConnections(callback) {
this._request({
cmd: 'NODE_SOCKET_GET_COUNT',
key: this.key
}, 'NODE_SOCKET_COUNT', function(err, msg) {
if (err) return callback(err);
callback(null, msg.count);
});
}
}
this.slave.once('disconnect', onclose);
this.slave.on('internalMessage', onreply);
};
SocketListSend.prototype.close = function close(callback) {
this._request({
cmd: 'NODE_SOCKET_NOTIFY_CLOSE',
key: this.key
}, 'NODE_SOCKET_ALL_CLOSED', callback);
};
SocketListSend.prototype.getConnections = function getConnections(callback) {
this._request({
cmd: 'NODE_SOCKET_GET_COUNT',
key: this.key
}, 'NODE_SOCKET_COUNT', function(err, msg) {
if (err) return callback(err);
callback(null, msg.count);
});
};
// This object keep track of the socket there are received
function SocketListReceive(slave, key) {
EventEmitter.call(this);
class SocketListReceive extends EventEmitter {
constructor(child, key) {
super();
this.connections = 0;
this.key = key;
this.slave = slave;
this.connections = 0;
this.key = key;
this.child = child;
function onempty(self) {
if (!self.slave.connected) return;
function onempty(self) {
if (!self.child.connected) return;
self.slave.send({
cmd: 'NODE_SOCKET_ALL_CLOSED',
key: self.key
self.child.send({
cmd: 'NODE_SOCKET_ALL_CLOSED',
key: self.key
});
}
this.child.on('internalMessage', (msg) => {
if (msg.key !== this.key) return;
if (msg.cmd === 'NODE_SOCKET_NOTIFY_CLOSE') {
// Already empty
if (this.connections === 0) return onempty(this);
// Wait for sockets to get closed
this.once('empty', onempty);
} else if (msg.cmd === 'NODE_SOCKET_GET_COUNT') {
if (!this.child.connected) return;
this.child.send({
cmd: 'NODE_SOCKET_COUNT',
key: this.key,
count: this.connections
});
}
});
}
this.slave.on('internalMessage', (msg) => {
if (msg.key !== this.key) return;
if (msg.cmd === 'NODE_SOCKET_NOTIFY_CLOSE') {
// Already empty
if (this.connections === 0) return onempty(this);
// Wait for sockets to get closed
this.once('empty', onempty);
} else if (msg.cmd === 'NODE_SOCKET_GET_COUNT') {
if (!this.slave.connected) return;
this.slave.send({
cmd: 'NODE_SOCKET_COUNT',
key: this.key,
count: this.connections
});
}
});
}
util.inherits(SocketListReceive, EventEmitter);
add(obj) {
this.connections++;
SocketListReceive.prototype.add = function(obj) {
this.connections++;
// Notify previous owner of socket about its state change
obj.socket.once('close', () => {
this.connections--;
// Notify previous owner of socket about its state change
obj.socket.once('close', () => {
this.connections--;
if (this.connections === 0) this.emit('empty', this);
});
}
}
if (this.connections === 0) this.emit('empty', this);
});
};
module.exports = {SocketListSend, SocketListReceive};

0 comments on commit 88c3f57

Please sign in to comment.