mirror of https://github.com/lukechilds/node.git
Browse Source
PR-URL: https://github.com/nodejs/node/pull/11406 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Michaël Zasso <targos@protonmail.com>v6
James M Snell
8 years ago
1 changed files with 80 additions and 79 deletions
@ -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}; |
|||
|
Loading…
Reference in new issue