diff options
author | Feross Aboukhadijeh <feross@feross.org> | 2016-04-21 09:10:32 +0300 |
---|---|---|
committer | Feross Aboukhadijeh <feross@feross.org> | 2016-04-21 09:10:32 +0300 |
commit | 3daee2c66cbf752b9e6e49b99492b8c1914a4a58 (patch) | |
tree | fbdd6c10f6a64674268dc77d1b3ccb4a37c44a65 /lib/tcp-pool.js | |
parent | 7a7c4a8b8c49f5c92b7c20ff439bc8614f7d607e (diff) |
BREAKING: Major cleanup
### Added
- `client.listening` property to signal whether TCP server is listening
for incoming
connections.
### Changed
- Merged `Swarm` class into `Torrent` object. Properties on
`torrent.swarm` (like
`torrent.swarm.wires`) now exist on `torrent` (e.g. `torrent.wires`).
- `torrent.addPeer` can no longer be called before the `infoHash` event
has been
emitted.
- Remove `torrent.on('listening')` event. Use `client.on('listening')`
instead.
- Remove support from `TCPPool` for listening on multiple ports. This
was not used by
WebTorrent and just added complexity. There is now a single `TCPPool`
instance for the
whole WebTorrent client.
- Deprecate: Do not use `client.download()` anymore. Use `client.add()`
instead.
- Deprecate: Do not use `torrent.swarm` anymore. Use `torrent` instead.
### Fixed
- When there is a `torrent.on('error')` listener, don't also emit
`client.on('error')`.
- Do not return existing torrent object when duplicate torrent is
added. Fire an
`'error'` event instead.
- Memory leak of `Torrent` object caused by `RarityMap`
- Memory leak of `Torrent` object caused by `TCPPool`
- `client.ratio` and `torrent.ratio` are now calculated as `uploaded /
received` instead
of `uploaded / downloaded`.
Diffstat (limited to 'lib/tcp-pool.js')
-rw-r--r-- | lib/tcp-pool.js | 190 |
1 files changed, 44 insertions, 146 deletions
diff --git a/lib/tcp-pool.js b/lib/tcp-pool.js index 7e95f6c..5e15d0a 100644 --- a/lib/tcp-pool.js +++ b/lib/tcp-pool.js @@ -7,12 +7,6 @@ var net = require('net') // browser exclude var Peer = require('./peer') /** - * Shared TCP pools; shared among all swarms - * @type {Object} port: number -> pool: TCPPool - */ -var tcpPools = {} - -/** * TCPPool * * A "TCP pool" allows multiple swarms to listen on the same TCP port and determines @@ -20,107 +14,35 @@ var tcpPools = {} * handshake that the remote peer sends. * * @param {number} port - * @param {string} hostname */ -function TCPPool (port, hostname) { +function TCPPool (client) { var self = this - - self.port = port - self.listening = false - self.swarms = {} // infoHash (hex) -> Swarm - - debug('new TCPPool (port: %s, hostname: %s)', port, hostname) - - // Save incoming conns so they can be destroyed if server is closed before the conn is - // passed off to a Swarm. - self.pendingConns = [] + debug('create tcp pool (port %s)', client.torrentPort) self.server = net.createServer() - self.server.on('connection', function (conn) { self._onConnection(conn) }) - self.server.on('error', function (err) { self._onError(err) }) - self.server.on('listening', function () { self._onListening() }) - self.server.listen(self.port, hostname) -} - -/** - * STATIC METHOD - * Add a swarm to a pool, creating a new pool if necessary. - * @param {Swarm} swarm - */ -TCPPool.addSwarm = function (swarm) { - var pool = tcpPools[swarm._port] - if (!pool) pool = tcpPools[swarm._port] = new TCPPool(swarm._port, swarm._hostname) - pool.addSwarm(swarm) - return pool -} - -/** - * STATIC METHOD - * Remove a swarm from its pool. - * @param {Swarm} swarm - */ -TCPPool.removeSwarm = function (swarm, cb) { - var pool = tcpPools[swarm._port] - if (!pool) return cb() - pool.removeSwarm(swarm) + self._client = client - if (Object.keys(pool.swarms).length === 0) pool.destroy(cb) - else process.nextTick(cb) -} + // Temporarily store incoming connections so they can be destroyed if the server is + // closed before the connection is passed off to a Torrent. + self._pendingConns = [] -/** - * STATIC METHOD - * When `Swarm.prototype.listen` is called without specifying a port, a reasonable - * default port must be chosen. If there already exists an active TCP pool, then return - * that pool's port so that TCP server can be re-used. Otherwise, return 0 so node will - * pick a free port. - * - * @return {number} port - */ -TCPPool.getDefaultListenPort = function (infoHash) { - for (var port in tcpPools) { - var pool = tcpPools[port] - if (!pool.swarms[infoHash]) return pool.port + self._onConnectionBound = function (conn) { + self._onConnection(conn) } - return 0 -} - -/** - * Add a swarm to this TCP pool. - * @param {Swarm} swarm - */ -TCPPool.prototype.addSwarm = function (swarm) { - var self = this - if (self.swarms[swarm.infoHash]) { - process.nextTick(function () { - swarm._onError(new Error( - 'There is already a swarm with info hash ' + swarm.infoHash + ' ' + - 'listening on port ' + swarm._port - )) - }) - return + self._onListening = function () { + self._client._onListening() } - self.swarms[swarm.infoHash] = swarm - - if (self.listening) { - process.nextTick(function () { - swarm._onListening(self.port) - }) + self._onError = function (err) { + self._client._destroy(err) } - debug('add swarm %s to tcp pool %s', swarm.infoHash, self.port) -} + self.server.on('connection', self._onConnectionBound) + self.server.on('listening', self._onListening) + self.server.on('error', self._onError) -/** - * Remove a swarm from this TCP pool. - * @param {Swarm} swarm - */ -TCPPool.prototype.removeSwarm = function (swarm) { - var self = this - debug('remove swarm %s from tcp pool %s', swarm.infoHash, self.port) - delete self.swarms[swarm.infoHash] + self.server.listen(client.torrentPort) } /** @@ -129,47 +51,28 @@ TCPPool.prototype.removeSwarm = function (swarm) { */ TCPPool.prototype.destroy = function (cb) { var self = this - debug('destroy tcp pool %s', self.port) + debug('destroy tcp pool') - self.listening = false + self.server.removeListener('connection', self._onConnectionBound) + self.server.removeListener('listening', self._onListening) + self.server.removeListener('error', self._onError) // Destroy all open connection objects so server can close gracefully without waiting // for connection timeout or remote peer to disconnect. - self.pendingConns.forEach(function (conn) { + self._pendingConns.forEach(function (conn) { + conn.on('error', noop) conn.destroy() }) - delete tcpPools[self.port] - try { self.server.close(cb) } catch (err) { if (cb) process.nextTick(cb) } -} - -TCPPool.prototype._onListening = function () { - var self = this - - // Fix for Docker Node image. Sometimes server.address() returns `null`. - // See issue: https://github.com/feross/bittorrent-swarm/pull/18 - var address = self.server.address() || { port: 0 } - var port = address.port - - debug('tcp pool listening on %s', port) - if (port !== self.port) { - // `port` was 0 when `listen` was called; update to the port that node selected - delete tcpPools[self.port] - self.port = port - tcpPools[self.port] = self - } - - self.listening = true - - for (var infoHash in self.swarms) { - self.swarms[infoHash]._onListening(self.port) - } + self.server = null + self._client = null + self._pendingConns = null } /** @@ -189,39 +92,34 @@ TCPPool.prototype._onConnection = function (conn) { return } - self.pendingConns.push(conn) - conn.once('close', removePendingConn) - - function removePendingConn () { - arrayRemove(self.pendingConns, self.pendingConns.indexOf(conn)) - } + self._pendingConns.push(conn) + conn.once('close', cleanupPending) var peer = Peer.createTCPIncomingPeer(conn) - peer.wire.once('handshake', function (infoHash, peerId) { - removePendingConn() - conn.removeListener('close', removePendingConn) + var wire = peer.wire + wire.once('handshake', onHandshake) - var swarm = self.swarms[infoHash] - if (swarm) { - peer.swarm = swarm - swarm._addIncomingPeer(peer) + function onHandshake (infoHash, peerId) { + cleanupPending() + + var torrent = self._client.get(infoHash) + if (torrent) { + peer.swarm = torrent + torrent._addIncomingPeer(peer) peer.onHandshake(infoHash, peerId) } else { - var err = new Error('Unexpected info hash ' + infoHash + ' from incoming peer ' + - peer.id + ': destroying peer') + var err = new Error( + 'Unexpected info hash ' + infoHash + ' from incoming peer ' + peer.id + ) peer.destroy(err) } - }) -} + } -TCPPool.prototype._onError = function (err) { - var self = this - self.destroy() - for (var infoHash in self.swarms) { - var swarm = self.swarms[infoHash] - self.removeSwarm(swarm) - swarm._onError(err) + function cleanupPending () { + conn.removeListener('close', cleanupPending) + wire.removeListener('handshake', onHandshake) + arrayRemove(self._pendingConns, self._pendingConns.indexOf(conn)) } } |