feat(pull): migration to pull-streams

src/listener.js

@@ -0,0 +1,129 @@
+'use strict'
+
+const multiaddr = require('multiaddr')
+const debug = require('debug')
+const log = debug('libp2p:tcp')
+const Connection = require('interface-connection').Connection
+const os = require('os')
+const contains = require('lodash.contains')
+const net = require('net')
+const toPull = require('stream-to-pull-stream')
+
+const getMultiaddr = require('./get-multiaddr')
+
+const IPFS_CODE = 421
+const CLOSE_TIMEOUT = 2000
+
+module.exports = (handler) => {
+  const listener = net.createServer((socket) => {
+    const s = toPull.duplex(socket)
+    s.getObservedAddrs = (cb) => {
+      return cb(null, [getMultiaddr(socket)])
+    }
+
+    const conn = new Connection(s)
+    handler(conn)
+  })
+
+  // Keep track of open connections to destroy in case of timeout
+  listener.__connections = {}
+  listener.on('connection', (socket) => {
+    const key = `${socket.remoteAddress}:${socket.remotePort}`
+    listener.__connections[key] = socket
+
+    socket.on('close', () => {
+      delete listener.__connections[key]
+    })
+  })
+
+  listener._close = listener.close
+  listener.close = (options, cb) => {
+    if (typeof options === 'function') {
+      cb = options
+      options = {}
+    }
+    cb = cb || (() => {})
+    options = options || {}
+
+    let closed = false
+    listener._close(cb)
+    listener.once('close', () => {
+      closed = true
+    })
+    setTimeout(() => {
+      if (closed) return
+
+      log('unable to close graciously, destroying conns')
+      Object.keys(listener.__connections).forEach((key) => {
+        log('destroying %s', key)
+        listener.__connections[key].destroy()
+      })
+    }, options.timeout || CLOSE_TIMEOUT)
+  }
+  let ipfsId
+  let listeningAddr
+
+  listener._listen = listener.listen
+  listener.listen = (ma, cb) => {
+    listeningAddr = ma
+    if (contains(ma.protoNames(), 'ipfs')) {
+      ipfsId = getIpfsId(ma)
+      listeningAddr = ma.decapsulate('ipfs')
+    }
+
+    const lOpts = listeningAddr.toOptions()
+    log('Listening on %s %s', lOpts.port, lOpts.host)
+    return listener._listen(lOpts.port, lOpts.host, cb)
+  }
+
+  listener.getAddrs = (cb) => {
+    const multiaddrs = []
+    const address = listener.address()
+
+    if (!address) {
+      return cb(new Error('Listener is not ready yet'))
+    }
+
+    // Because TCP will only return the IPv6 version
+    // we need to capture from the passed multiaddr
+    if (listeningAddr.toString().indexOf('ip4') !== -1) {
+      let m = listeningAddr.decapsulate('tcp')
+      m = m.encapsulate('/tcp/' + address.port)
+      if (ipfsId) {
+        m = m.encapsulate('/ipfs/' + ipfsId)
+      }
+
+      if (m.toString().indexOf('0.0.0.0') !== -1) {
+        const netInterfaces = os.networkInterfaces()
+        Object.keys(netInterfaces).forEach((niKey) => {
+          netInterfaces[niKey].forEach((ni) => {
+            if (ni.family === 'IPv4') {
+              multiaddrs.push(multiaddr(m.toString().replace('0.0.0.0', ni.address)))
+            }
+          })
+        })
+      } else {
+        multiaddrs.push(m)
+      }
+    }
+
+    if (address.family === 'IPv6') {
+      let ma = multiaddr('/ip6/' + address.address + '/tcp/' + address.port)
+      if (ipfsId) {
+        ma = ma.encapsulate('/ipfs/' + ipfsId)
+      }
+
+      multiaddrs.push(ma)
+    }
+
+    cb(null, multiaddrs)
+  }
+
+  return listener
+}
+
+function getIpfsId (ma) {
+  return ma.stringTuples().filter((tuple) => {
+    return tuple[0] === IPFS_CODE
+  })[0][1]
+}