Support async createSwarm() and getSwarm()

Fix #95.

Make server.getSwarm() and server.createSwarm() into async functions
that take a callback.
This commit is contained in:
Feross Aboukhadijeh 2016-01-03 19:50:23 +01:00
parent 9d4c404dc6
commit b5096e91c3
3 changed files with 118 additions and 90 deletions

View File

@ -4,7 +4,7 @@ var debug = require('debug')('bittorrent-tracker')
var randomIterate = require('random-iterate') var randomIterate = require('random-iterate')
// Regard this as the default implementation of an interface that you // Regard this as the default implementation of an interface that you
// need to support when overriding Server.getSwarm() // need to support when overriding Server.createSwarm() and Server.getSwarm()
function Swarm (infoHash, server) { function Swarm (infoHash, server) {
this.peers = {} this.peers = {}
this.complete = 0 this.complete = 0

View File

@ -196,17 +196,23 @@ Server.prototype.close = function (cb) {
else cb(null) else cb(null)
} }
Server.prototype.createSwarm = function (infoHash) { Server.prototype.createSwarm = function (infoHash, cb) {
var self = this var self = this
if (Buffer.isBuffer(infoHash)) infoHash = infoHash.toString('hex') if (Buffer.isBuffer(infoHash)) infoHash = infoHash.toString('hex')
process.nextTick(function () {
var swarm = self.torrents[infoHash] = new Swarm(infoHash, self) var swarm = self.torrents[infoHash] = new Swarm(infoHash, self)
return swarm cb(null, swarm)
})
} }
Server.prototype.getSwarm = function (infoHash) { Server.prototype.getSwarm = function (infoHash, cb) {
var self = this var self = this
if (Buffer.isBuffer(infoHash)) infoHash = infoHash.toString('hex') if (Buffer.isBuffer(infoHash)) infoHash = infoHash.toString('hex')
return self.torrents[infoHash]
process.nextTick(function () {
cb(null, self.torrents[infoHash])
})
} }
Server.prototype.onHttpRequest = function (req, res, opts) { Server.prototype.onHttpRequest = function (req, res, opts) {
@ -358,7 +364,8 @@ Server.prototype._onWebSocketRequest = function (socket, params) {
if (params.answer) { if (params.answer) {
debug('got answer %s from %s', JSON.stringify(params.answer), params.peer_id) debug('got answer %s from %s', JSON.stringify(params.answer), params.peer_id)
var swarm = self.getSwarm(params.info_hash) self.getSwarm(params.info_hash, function (err, swarm) {
if (err) return self.emit('warning', err)
if (!swarm) { if (!swarm) {
return self.emit('warning', new Error('no swarm with that `info_hash`')) return self.emit('warning', new Error('no swarm with that `info_hash`'))
} }
@ -374,11 +381,19 @@ Server.prototype._onWebSocketRequest = function (socket, params) {
info_hash: common.hexToBinary(params.info_hash) info_hash: common.hexToBinary(params.info_hash)
}), toPeer.socket.onSend) }), toPeer.socket.onSend)
debug('sent answer to %s from %s', toPeer.peerId, params.peer_id) debug('sent answer to %s from %s', toPeer.peerId, params.peer_id)
done()
})
} else {
done()
} }
function done () {
// emit event once the announce is fully "processed"
if (params.action === common.ACTIONS.ANNOUNCE) { if (params.action === common.ACTIONS.ANNOUNCE) {
self.emit(common.EVENT_NAMES[params.event], params.peer_id, params) self.emit(common.EVENT_NAMES[params.event], params.peer_id, params)
} }
}
}) })
} }
@ -398,9 +413,14 @@ Server.prototype._onRequest = function (params, cb) {
Server.prototype._onAnnounce = function (params, cb) { Server.prototype._onAnnounce = function (params, cb) {
var self = this var self = this
var swarm = self.getSwarm(params.info_hash) self.getSwarm(params.info_hash, function (err, swarm) {
if (swarm) announce() if (err) return cb(err)
else createSwarm() if (swarm) {
announce(swarm)
} else {
createSwarm()
}
})
function createSwarm () { function createSwarm () {
if (self._filter) { if (self._filter) {
@ -410,17 +430,21 @@ Server.prototype._onAnnounce = function (params, cb) {
} else if (!allowed) { } else if (!allowed) {
cb(new Error('disallowed info_hash')) cb(new Error('disallowed info_hash'))
} else { } else {
swarm = self.createSwarm(params.info_hash) self.createSwarm(params.info_hash, function (err, swarm) {
announce() if (err) return cb(err)
announce(swarm)
})
} }
}) })
} else { } else {
swarm = self.createSwarm(params.info_hash) self.createSwarm(params.info_hash, function (err, swarm) {
announce() if (err) return cb(err)
announce(swarm)
})
} }
} }
function announce () { function announce (swarm) {
if (!params.event || params.event === 'empty') params.event = 'update' if (!params.event || params.event === 'empty') params.event = 'update'
swarm.announce(params, function (err, response) { swarm.announce(params, function (err, response) {
if (err) return cb(err) if (err) return cb(err)
@ -470,7 +494,8 @@ Server.prototype._onScrape = function (params, cb) {
series(params.info_hash.map(function (infoHash) { series(params.info_hash.map(function (infoHash) {
return function (cb) { return function (cb) {
var swarm = self.getSwarm(infoHash) self.getSwarm(infoHash, function (err, swarm) {
if (err) return cb(err)
if (swarm) { if (swarm) {
swarm.scrape(params, function (err, scrapeInfo) { swarm.scrape(params, function (err, scrapeInfo) {
if (err) return cb(err) if (err) return cb(err)
@ -483,6 +508,7 @@ Server.prototype._onScrape = function (params, cb) {
} else { } else {
cb(null, { infoHash: infoHash, complete: 0, incomplete: 0 }) cb(null, { infoHash: infoHash, complete: 0, incomplete: 0 })
} }
})
} }
}), function (err, results) { }), function (err, results) {
if (err) return cb(err) if (err) return cb(err)

View File

@ -8,7 +8,7 @@ var peerId2 = new Buffer('12345678901234567890')
var torrentLength = 50000 var torrentLength = 50000
function serverTest (t, serverType, serverFamily) { function serverTest (t, serverType, serverFamily) {
t.plan(25) t.plan(26)
var opts = serverType === 'http' ? { udp: false, ws: false } : { http: false, ws: false } var opts = serverType === 'http' ? { udp: false, ws: false } : { http: false, ws: false }
var server = new Server(opts) var server = new Server(opts)
@ -49,7 +49,8 @@ function serverTest (t, serverType, serverFamily) {
t.equal(data.complete, 0) t.equal(data.complete, 0)
t.equal(data.incomplete, 1) t.equal(data.incomplete, 1)
var swarm = server.getSwarm(infoHash) server.getSwarm(infoHash, function (err, swarm) {
t.error(err)
t.equal(Object.keys(server.torrents).length, 1) t.equal(Object.keys(server.torrents).length, 1)
t.equal(swarm.complete, 0) t.equal(swarm.complete, 0)
@ -115,6 +116,7 @@ function serverTest (t, serverType, serverFamily) {
}) })
}) })
}) })
})
} }
test('http ipv4 server', function (t) { test('http ipv4 server', function (t) {