udp/ws clients: add destroy functions

Fixes webtorrent#75

Author: feross

client.js

@@ -227,7 +227,7 @@ Client.prototype.destroy = function () {
   debug('destroy')
 
   self._trackers.forEach(function (tracker) {
-    if (tracker.destroy) tracker.destroy()
+    tracker.destroy()
     tracker.setInterval(0) // stop announcing on intervals
   })
   self._trackers = []

lib/udp-tracker.js

@@ -11,6 +11,8 @@ var url = require('url')
 
 var common = require('./common')
 
+var TIMEOUT = 15000
+
 inherits(UDPTracker, EventEmitter)
 
 /**
@@ -31,6 +33,7 @@ function UDPTracker (client, announceUrl, opts) {
   self._announceUrl = announceUrl
   self._intervalMs = self.client._intervalMs // use client interval initially
   self._interval = null
+  self._cleanupFns = []
 }
 
 UDPTracker.prototype.announce = function (opts) {
@@ -44,60 +47,90 @@ UDPTracker.prototype.scrape = function (opts) {
   self._request(opts) // udp scrape uses same announce url
 }
 
+// TODO: Improve this interface
+UDPTracker.prototype.setInterval = function (intervalMs) {
+  var self = this
+  clearInterval(self._interval)
+
+  self._intervalMs = intervalMs
+  if (intervalMs) {
+    // HACK
+    var update = self.announce.bind(self, self.client._defaultAnnounceOpts())
+    self._interval = setInterval(update, self._intervalMs)
+  }
+}
+
+UDPTracker.prototype.destroy = function () {
+  var self = this
+  if (self.destroyed) return
+  self.destroyed = true
+
+  self._cleanupFns.slice(0).forEach(function (cleanup) {
+    cleanup()
+  })
+  self._cleanupFns = []
+}
+
 UDPTracker.prototype._request = function (opts) {
   var self = this
   if (!opts) opts = {}
   var parsedUrl = url.parse(self._announceUrl)
-  var socket = dgram.createSocket('udp4')
   var transactionId = genTransactionId()
+  var socket = dgram.createSocket('udp4')
+
+  var cleanup = function () {
+    self._cleanupFns.splice(self._cleanupFns.indexOf(cleanup), 1)
+    if (timeout) {
+      clearTimeout(timeout)
+      timeout = null
+    }
+    socket.removeListener('error', onError)
+    socket.removeListener('message', onSocketMessage)
+    try { socket.close() } catch (err) {}
+    socket = null
+  }
+  self._cleanupFns.push(cleanup)
 
   // does not matter if `stopped` event arrives, so supress errors & cleanup after timeout
+  var ms = opts.event === 'stopped' ? TIMEOUT / 10 : TIMEOUT
   var timeout = setTimeout(function () {
     timeout = null
     cleanup()
     if (opts.event !== 'stopped') {
-      error('tracker request timed out')
+      onError(new Error('tracker request timed out'))
     }
-  }, opts.event === 'stopped' ? 1500 : 15000)
-
-  if (timeout && timeout.unref) {
-    timeout.unref()
-  }
+  }, ms)
+  if (timeout.unref) timeout.unref()
 
   send(Buffer.concat([
     common.CONNECTION_ID,
     common.toUInt32(common.ACTIONS.CONNECT),
     transactionId
   ]))
 
-  socket.on('error', error)
+  socket.on('error', onError)
+  socket.on('message', onSocketMessage)
 
-  socket.on('message', function (msg) {
+  function onSocketMessage (msg) {
+    if (self.destroyed) return
     if (msg.length < 8 || msg.readUInt32BE(4) !== transactionId.readUInt32BE(0)) {
-      return error('tracker sent invalid transaction id')
+      return onError(new Error('tracker sent invalid transaction id'))
     }
 
     var action = msg.readUInt32BE(0)
-    debug(self._announceUrl + ' UDP response, action ' + action)
+    debug('UDP response %s, action %s', self._announceUrl, action)
     switch (action) {
       case 0: // handshake
-        if (msg.length < 16) {
-          return error('invalid udp handshake')
-        }
+        if (msg.length < 16) return onError(new Error('invalid udp handshake'))
 
-        if (opts._scrape) {
-          scrape(msg.slice(8, 16))
-        } else {
-          announce(msg.slice(8, 16), opts)
-        }
+        if (opts._scrape) scrape(msg.slice(8, 16))
+        else announce(msg.slice(8, 16), opts)
 
         return
 
       case 1: // announce
         cleanup()
-        if (msg.length < 20) {
-          return error('invalid announce message')
-        }
+        if (msg.length < 20) return onError(new Error('invalid announce message'))
 
         var interval = msg.readUInt32BE(8)
         if (interval && !self._opts.interval && self._intervalMs !== 0) {
@@ -126,7 +159,7 @@ UDPTracker.prototype._request = function (opts) {
       case 2: // scrape
         cleanup()
         if (msg.length < 20 || (msg.length - 8) % 12 !== 0) {
-          return error('invalid scrape message')
+          return onError(new Error('invalid scrape message'))
         }
         var infoHashes = (Array.isArray(opts.infoHash) && opts.infoHash.length > 0)
           ? opts.infoHash.map(function (infoHash) { return infoHash.toString('hex') })
@@ -145,37 +178,29 @@ UDPTracker.prototype._request = function (opts) {
 
       case 3: // error
         cleanup()
-        if (msg.length < 8) {
-          return error('invalid error message')
-        }
+        if (msg.length < 8) return onError(new Error('invalid error message'))
         self.client.emit('warning', new Error(msg.slice(8).toString()))
         break
 
       default:
-        error('tracker sent invalid action')
+        onError(new Error('tracker sent invalid action'))
         break
     }
-  })
-
-  function send (message) {
-    if (!parsedUrl.port) {
-      parsedUrl.port = 80
-    }
-    socket.send(message, 0, message.length, parsedUrl.port, parsedUrl.hostname)
   }
 
-  function error (message) {
-    // errors will often happen if a tracker is offline, so don't treat it as fatal
-    self.client.emit('warning', new Error(message + ' (' + self._announceUrl + ')'))
+  function onError (err) {
+    if (self.destroyed) return
     cleanup()
+    if (err.message) err.message += ' (' + self._announceUrl + ')'
+    // errors will often happen if a tracker is offline, so don't treat it as fatal
+    self.client.emit('warning', err)
   }
 
-  function cleanup () {
-    if (timeout) {
-      clearTimeout(timeout)
-      timeout = null
+  function send (message) {
+    if (!parsedUrl.port) {
+      parsedUrl.port = 80
     }
-    try { socket.close() } catch (err) {}
+    socket.send(message, 0, message.length, parsedUrl.port, parsedUrl.hostname)
   }
 
   function announce (connectionId, opts) {
@@ -215,19 +240,6 @@ UDPTracker.prototype._request = function (opts) {
   }
 }
 
-// TODO: Improve this interface
-UDPTracker.prototype.setInterval = function (intervalMs) {
-  var self = this
-  clearInterval(self._interval)
-
-  self._intervalMs = intervalMs
-  if (intervalMs) {
-    // HACK
-    var update = self.announce.bind(self, self.client._defaultAnnounceOpts())
-    self._interval = setInterval(update, self._intervalMs)
-  }
-}
-
 function genTransactionId () {
   return new Buffer(hat(32), 'hex')
 }

lib/websocket-tracker.js

@@ -1,4 +1,4 @@
-// TODO: destroy the websocket
+// TODO: cleanup unused Peers when tracker doesn't respond with matches
 
 module.exports = WebSocketTracker
 
@@ -11,10 +11,6 @@ var Socket = require('simple-websocket')
 
 var common = require('./common')
 
-// It turns out that you can't open multiple websockets to the same server within one
-// browser tab, so let's reuse them.
-var socketPool = {}
-
 inherits(WebSocketTracker, EventEmitter)
 
 function WebSocketTracker (client, announceUrl, opts) {
@@ -31,11 +27,12 @@ function WebSocketTracker (client, announceUrl, opts) {
   self._intervalMs = self.client._intervalMs // use client interval initially
   self._interval = null
 
-  if (socketPool[announceUrl]) self._socket = socketPool[announceUrl]
-  else self._socket = socketPool[announceUrl] = new Socket(announceUrl)
+  self._onSocketErrorBound = self._onSocketError.bind(self)
+  self._onSocketDataBound = self._onSocketData.bind(self)
 
-  self._socket.on('error', self._onSocketError.bind(self))
-  self._socket.on('data', self._onSocketData.bind(self))
+  self._socket = new Socket(announceUrl + '?' + hat(40))
+  self._socket.on('error', self._onSocketErrorBound)
+  self._socket.on('data', self._onSocketDataBound)
 }
 
 WebSocketTracker.prototype.announce = function (opts) {
@@ -79,13 +76,24 @@ WebSocketTracker.prototype.setInterval = function (intervalMs) {
   }
 }
 
+WebSocketTracker.prototype.destroy = function () {
+  var self = this
+  if (self.destroyed) return
+  self.destroyed = true
+  self._socket.removeListener('error', self._onSocketErrorBound)
+  self._socket.removeListener('data', self._onSocketDataBound)
+  self._socket.close()
+}
+
 WebSocketTracker.prototype._onSocketError = function (err) {
   var self = this
+  if (self.destroyed) return
   self.client.emit('error', err)
 }
 
 WebSocketTracker.prototype._onSocketData = function (data) {
   var self = this
+  if (self.destroyed) return
 
   if (!(typeof data === 'object' && data !== null)) {
     return self.client.emit('warning', new Error('Invalid tracker response'))
@@ -161,6 +169,8 @@ WebSocketTracker.prototype._onSocketData = function (data) {
 
 WebSocketTracker.prototype._send = function (params) {
   var self = this
+  if (self.destroyed) return
+
   var message = JSON.stringify(params)
   debug('send %s', message)
   self._socket.send(message)