chore: address review

Author: vasco-santos

package.json

@@ -46,7 +46,8 @@
     "libp2p-utils": "~0.1.0",
     "mafmt": "^7.0.0",
     "multiaddr": "^7.1.0",
-    "multiaddr-to-uri": "^5.0.0"
+    "multiaddr-to-uri": "^5.0.0",
+    "p-timeout": "^3.2.0"
   },
   "devDependencies": {
     "abort-controller": "^3.0.0",

src/index.js

@@ -37,8 +37,8 @@ class WebSockets {
   async dial (ma, options = {}) {
     log('dialing %s', ma)
 
-    const stream = await this._connect(ma, options)
-    const maConn = toConnection(stream, { socket: stream.socket, remoteAddr: ma, signal: options.signal })
+    const socket = await this._connect(ma, options)
+    const maConn = toConnection(socket, { remoteAddr: ma, signal: options.signal })
     log('new outbound connection %s', maConn.remoteAddr)
 
     const conn = await this._upgrader.upgradeOutbound(maConn)
@@ -51,7 +51,7 @@ class WebSockets {
    * @param {Multiaddr} ma
    * @param {object} [options]
    * @param {AbortSignal} [options.signal] Used to abort dial requests
-   * @returns {Promise<Socket>} Resolves a TCP Socket
+   * @returns {Promise<WebSocket>} Resolves a extended duplex iterable on top of a WebSocket
    */
   async _connect (ma, options = {}) {
     if (options.signal && options.signal.aborted) {
@@ -101,12 +101,11 @@ class WebSockets {
    * @param {function (Connection)} handler
    * @returns {Listener} A Websockets listener
    */
-  createListener (options, handler) {
+  createListener (options = {}, handler) {
     if (typeof options === 'function') {
       handler = options
       options = {}
     }
-    options = options || {}
 
     return createListener({ handler, upgrader: this._upgrader }, options)
   }

src/listener.js

@@ -7,16 +7,13 @@ const { createServer } = require('it-ws')
 
 const log = require('debug')('libp2p:websockets:listener')
 
-const { CODE_P2P } = require('./constants')
 const toConnection = require('./socket-to-conn')
 
 module.exports = ({ handler, upgrader }, options = {}) => {
   const listener = new EventEmitter()
 
-  const server = createServer(options, async (stream, req) => {
-    const maConn = toConnection(stream, {
-      socket: req.socket
-    })
+  const server = createServer(options, async (stream) => {
+    const maConn = toConnection(stream)
 
     log('new inbound connection %s', maConn.remoteAddr)
 
@@ -37,7 +34,7 @@ module.exports = ({ handler, upgrader }, options = {}) => {
   // Keep track of open connections to destroy in case of timeout
   server.__connections = []
 
-  let peerId, listeningMultiaddr
+  let listeningMultiaddr
 
   listener.close = () => {
     server.__connections.forEach(maConn => maConn.close())
@@ -46,11 +43,6 @@ module.exports = ({ handler, upgrader }, options = {}) => {
 
   listener.listen = (ma) => {
     listeningMultiaddr = ma
-    peerId = listeningMultiaddr.getPeerId()
-
-    if (peerId) {
-      ma = ma.decapsulateCode(CODE_P2P)
-    }
 
     return server.listen(ma.toOptions())
   }
@@ -96,10 +88,4 @@ module.exports = ({ handler, upgrader }, options = {}) => {
 
 function trackConn (server, maConn) {
   server.__connections.push(maConn)
-
-  const untrackConn = () => {
-    server.__connections = server.__connections.filter(c => c !== maConn)
-  }
-
-  maConn.conn.once('close', untrackConn)
 }

src/socket-to-conn.js

@@ -4,63 +4,56 @@ const abortable = require('abortable-iterator')
 const { CLOSE_TIMEOUT } = require('./constants')
 const toMultiaddr = require('libp2p-utils/src/ip-port-to-multiaddr')
 
-const log = require('debug')('libp2p:websockets:socket')
+const pTimeout = require('p-timeout')
+
+const debug = require('debug')
+const log = debug('libp2p:websockets:socket')
+log.error = debug('libp2p:websockets:socket:error')
 
 // Convert a stream into a MultiaddrConnection
 // https://github.com/libp2p/interface-transport#multiaddrconnection
-module.exports = (stream, options = {}) => {
-  const socket = options.socket
-
+module.exports = (socket, options = {}) => {
   const maConn = {
     async sink (source) {
       if (options.signal) {
         source = abortable(source, options.signal)
       }
 
       try {
-        await stream.sink(source)
+        await socket.sink(source)
       } catch (err) {
-        // Re-throw non-aborted errors
-        if (err.type !== 'aborted') throw err
-        // Otherwise, this is fine...
-        await stream.close()
+        if (err.type !== 'aborted') {
+          log.error(err)
+        }
       }
     },
 
-    source: options.signal ? abortable(stream.source, options.signal) : stream.source,
+    source: options.signal ? abortable(socket.source, options.signal) : socket.source,
 
     conn: socket,
 
-    localAddr: undefined,
+    localAddr: options.localAddr || (socket.localAddress && socket.localPort
+      ? toMultiaddr(socket.localAddress, socket.localPort) : undefined),
 
     // If the remote address was passed, use it - it may have the peer ID encapsulated
-    remoteAddr: options.remoteAddr || toMultiaddr(stream.remoteAddress, stream.remotePort),
+    remoteAddr: options.remoteAddr || toMultiaddr(socket.remoteAddress, socket.remotePort),
 
     timeline: { open: Date.now() },
 
-    close () {
-      return new Promise(async (resolve) => { // eslint-disable-line no-async-promise-executor
-        const start = Date.now()
-
-        // Attempt to end the socket. If it takes longer to close than the
-        // timeout, destroy it manually.
-        const timeout = setTimeout(() => {
-          const { host, port } = maConn.remoteAddr.toOptions()
-          log('timeout closing socket to %s:%s after %dms, destroying it manually',
-            host, port, Date.now() - start)
-
-          socket.terminate()
-          maConn.timeline.close = Date.now()
-          return resolve()
-        }, CLOSE_TIMEOUT)
+    async close () {
+      const start = Date.now()
 
-        await stream.close()
+      try {
+        await pTimeout(socket.close(), CLOSE_TIMEOUT)
+      } catch (err) {
+        const { host, port } = maConn.remoteAddr.toOptions()
+        log('timeout closing socket to %s:%s after %dms, destroying it manually',
+          host, port, Date.now() - start)
 
-        clearTimeout(timeout)
+        socket.destroy()
+      } finally {
         maConn.timeline.close = Date.now()
-
-        resolve()
-      })
+      }
     }
   }
 

test/browser.js

@@ -47,7 +47,7 @@ describe('libp2p-websockets', () => {
     })
 
     it('many writes', async function () {
-      this.timeout(100000)
+      this.timeout(10000)
       const s = goodbye({
         source: pipe(
           {

test/node.js

@@ -11,9 +11,8 @@ const expect = chai.expect
 chai.use(dirtyChai)
 const multiaddr = require('multiaddr')
 const goodbye = require('it-goodbye')
-const { collect, consume } = require('streaming-iterables')
+const { collect } = require('streaming-iterables')
 const pipe = require('it-pipe')
-const AbortController = require('abort-controller')
 
 const WS = require('../src')
 
@@ -217,33 +216,31 @@ describe('dial', () => {
       expect(result).to.be.eql([Buffer.from('hey')])
     })
 
-    it('should be abortable after connect', async () => {
-      const controller = new AbortController()
-      const conn = await ws.dial(ma, { signal: controller.signal })
-      const s = goodbye({
-        source: {
-          [Symbol.asyncIterator] () {
-            return this
-          },
-          next () {
-            return new Promise(resolve => {
-              setTimeout(() => resolve(Math.random()), 1000)
-            })
-          }
-        },
-        sink: consume
-      })
+    it('should resolve port 0', async () => {
+      const ma = multiaddr('/ip4/127.0.0.1/tcp/0/ws')
+      const ws = new WS({ upgrader: mockUpgrader })
+
+      // Create a Promise that resolves when a connection is handled
+      let handled
+      const handlerPromise = new Promise(resolve => { handled = resolve })
+      const handler = conn => handled(conn)
+
+      const listener = ws.createListener(handler)
+
+      // Listen on the multiaddr
+      await listener.listen(ma)
 
-      setTimeout(() => controller.abort(), 500)
+      const localAddrs = listener.getAddrs()
+      expect(localAddrs.length).to.equal(1)
 
-      try {
-        await pipe(s, conn, s)
-      } catch (err) {
-        expect(err.type).to.equal('aborted')
-        return
-      }
+      // Dial to that address
+      await ws.dial(localAddrs[0])
 
-      throw new Error('connection was not aborted')
+      // Wait for the incoming dial to be handled
+      await handlerPromise
+
+      // close the listener
+      await listener.close()
     })
   })
 
@@ -446,36 +443,3 @@ describe('filter addrs', () => {
     done()
   })
 })
-
-describe.skip('valid localAddr and remoteAddr', () => {
-  const ma = multiaddr('/ip4/127.0.0.1/tcp/0/ws')
-
-  it('should resolve port 0', async () => {
-    const ws = new WS({ upgrader: mockUpgrader })
-
-    // Create a Promise that resolves when a connection is handled
-    let handled
-    const handlerPromise = new Promise(resolve => { handled = resolve })
-    const handler = conn => handled(conn)
-
-    const listener = ws.createListener(handler)
-
-    // Listen on the multiaddr
-    await listener.listen(ma)
-
-    const localAddrs = listener.getAddrs()
-    expect(localAddrs.length).to.equal(1)
-
-    // Dial to that address
-    const dialerConn = await ws.dial(localAddrs[0])
-
-    // Wait for the incoming dial to be handled
-    const listenerConn = await handlerPromise
-
-    // close the listener
-    await listener.close()
-
-    expect(dialerConn.localAddr.toString()).to.equal(listenerConn.remoteAddr.toString())
-    expect(dialerConn.remoteAddr.toString()).to.equal(listenerConn.localAddr.toString())
-  })
-})