refactor: wip switch to it-ws and async iterators

License: MIT
Signed-off-by: Alan Shaw <alan.shaw@protocol.ai>

package.json

@@ -43,9 +43,9 @@
     "class-is": "^1.1.0",
     "debug": "^4.1.1",
     "interface-connection": "~0.3.2",
+    "it-ws": "^1.0.0",
     "mafmt": "^6.0.4",
-    "multiaddr-to-uri": "^4.0.1",
-    "pull-ws": "hugomrdias/pull-ws#fix/bundle-size"
+    "multiaddr-to-uri": "^4.0.1"
   },
   "devDependencies": {
     "aegir": "^18.0.3",

src/index.js

@@ -1,47 +1,24 @@
 'use strict'
 
-const connect = require('pull-ws/client')
+const connect = require('it-ws/client')
 const mafmt = require('mafmt')
 const withIs = require('class-is')
-const Connection = require('interface-connection').Connection
-
 const toUri = require('multiaddr-to-uri')
-const debug = require('debug')
-const log = debug('libp2p:websockets:dialer')
+const log = require('debug')('libp2p:websockets:transport')
 
 const createListener = require('./listener')
 
 class WebSockets {
-  dial (ma, options, callback) {
-    if (typeof options === 'function') {
-      callback = options
-      options = {}
-    }
-
-    callback = callback || function () { }
-
+  async dial (ma, options) {
+    log('dialing %s', ma)
     const url = toUri(ma)
-    log('dialing %s', url)
-    const socket = connect(url, {
-      binary: true,
-      onConnect: (err) => {
-        callback(err)
-      }
-    })
-
-    const conn = new Connection(socket)
-    conn.getObservedAddrs = (cb) => cb(null, [ma])
-    conn.close = (cb) => socket.close(cb)
-
-    return conn
+    const socket = connect(url, { binary: true })
+    socket.getObservedAddrs = () => [ma]
+    log('connected %s', ma)
+    return socket
   }
 
   createListener (options, handler) {
-    if (typeof options === 'function') {
-      handler = options
-      options = {}
-    }
-
     return createListener(options, handler)
   }
 

src/listener.js

@@ -1,43 +1,35 @@
 'use strict'
 
-const Connection = require('interface-connection').Connection
 const multiaddr = require('multiaddr')
 const os = require('os')
 
-function noop () {}
-
-const createServer = require('pull-ws/server') || noop
+const createServer = require('it-ws/server')
 
 module.exports = (options, handler) => {
-  const listener = createServer(options, (socket) => {
-    socket.getObservedAddrs = (callback) => {
-      // TODO research if we can reuse the address in anyway
-      return callback(null, [])
-    }
-
-    handler(new Connection(socket))
+  const server = createServer(options, socket => {
+    socket.getObservedAddrs = () => []
+    handler(socket)
   })
 
   let listeningMultiaddr
 
-  listener._listen = listener.listen
-  listener.listen = (ma, callback) => {
-    callback = callback || noop
+  const listen = server.listen
+  server.listen = ma => {
     listeningMultiaddr = ma
 
     if (ma.protoNames().includes('ipfs')) {
       ma = ma.decapsulate('ipfs')
     }
 
-    listener._listen(ma.toOptions(), callback)
+    return listen(ma.toOptions())
   }
 
-  listener.getAddrs = (callback) => {
+  server.getAddrs = async () => {
     const multiaddrs = []
-    const address = listener.address()
+    const address = server.address()
 
     if (!address) {
-      return callback(new Error('Listener is not ready yet'))
+      throw new Error('Listener is not ready yet')
     }
 
     let ipfsId = listeningMultiaddr.getPeerId()
@@ -65,8 +57,8 @@ module.exports = (options, handler) => {
       }
     }
 
-    callback(null, multiaddrs)
+    return multiaddrs
   }
 
-  return listener
+  return server
 }

test/node.js

@@ -30,19 +30,18 @@ describe('listen', () => {
       ws = new WS()
     })
 
-    it('listen, check for callback', (done) => {
+    it('listen, check for promise', async () => {
       const listener = ws.createListener((conn) => { })
-
-      listener.listen(ma, () => {
-        listener.close(done)
-      })
+      await listener.listen(ma)
+      await listener.close()
     })
 
     it('listen, check for listening event', (done) => {
       const listener = ws.createListener((conn) => { })
 
-      listener.on('listening', () => {
-        listener.close(done)
+      listener.on('listening', async () => {
+        await listener.close()
+        done()
       })
 
       listener.listen(ma)
@@ -59,14 +58,12 @@ describe('listen', () => {
       listener.listen(ma)
     })
 
-    it('listen on addr with /ipfs/QmHASH', (done) => {
+    it('listen on addr with /ipfs/QmHASH', async () => {
       const ma = multiaddr('/ip4/127.0.0.1/tcp/9090/ws/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
-
       const listener = ws.createListener((conn) => { })
 
-      listener.listen(ma, () => {
-        listener.close(done)
-      })
+      await listener.listen(ma)
+      await listener.close()
     })
 
     it.skip('close listener with connections, through timeout', (done) => {
@@ -82,73 +79,53 @@ describe('listen', () => {
       // TODO 0.0.0.0 not supported yet
     })
 
-    it('getAddrs', (done) => {
-      const listener = ws.createListener((conn) => {
-      })
-      listener.listen(ma, () => {
-        listener.getAddrs((err, addrs) => {
-          expect(err).to.not.exist()
+    it('getAddrs', async () => {
+      const listener = ws.createListener((conn) => { })
+      await listener.listen(ma)
+      const addrs = await listener.getAddrs()
       expect(addrs.length).to.equal(1)
       expect(addrs[0]).to.deep.equal(ma)
-          listener.close(done)
-        })
-      })
+      await listener.close()
     })
 
-    it('getAddrs on port 0 listen', (done) => {
+    it('getAddrs on port 0 listen', async () => {
       const addr = multiaddr(`/ip4/127.0.0.1/tcp/0/ws`)
-      const listener = ws.createListener((conn) => {
-      })
-      listener.listen(addr, () => {
-        listener.getAddrs((err, addrs) => {
-          expect(err).to.not.exist()
+      const listener = ws.createListener((conn) => { })
+      await listener.listen(addr)
+      const addrs = await listener.getAddrs()
       expect(addrs.length).to.equal(1)
       expect(addrs.map((a) => a.toOptions().port)).to.not.include('0')
-          listener.close(done)
-        })
-      })
+      await listener.close()
     })
 
-    it('getAddrs from listening on 0.0.0.0', (done) => {
+    it('getAddrs from listening on 0.0.0.0', async () => {
       const addr = multiaddr(`/ip4/0.0.0.0/tcp/9003/ws`)
-      const listener = ws.createListener((conn) => {
-      })
-      listener.listen(addr, () => {
-        listener.getAddrs((err, addrs) => {
-          expect(err).to.not.exist()
+      const listener = ws.createListener((conn) => { })
+      await listener.listen(addr)
+      const addrs = await listener.getAddrs()
       expect(addrs.map((a) => a.toOptions().host)).to.not.include('0.0.0.0')
-          listener.close(done)
-        })
-      })
+      await listener.close()
     })
 
-    it('getAddrs from listening on 0.0.0.0 and port 0', (done) => {
+    it('getAddrs from listening on 0.0.0.0 and port 0', async () => {
       const addr = multiaddr(`/ip4/0.0.0.0/tcp/0/ws`)
-      const listener = ws.createListener((conn) => {
-      })
-      listener.listen(addr, () => {
-        listener.getAddrs((err, addrs) => {
-          expect(err).to.not.exist()
+      const listener = ws.createListener((conn) => { })
+      await listener.listen(addr)
+      const addrs = await listener.getAddrs()
       expect(addrs.map((a) => a.toOptions().host)).to.not.include('0.0.0.0')
       expect(addrs.map((a) => a.toOptions().port)).to.not.include('0')
-          listener.close(done)
-        })
-      })
+      await listener.close()
     })
 
-    it('getAddrs preserves IPFS Id', (done) => {
+    it('getAddrs preserves IPFS Id', async () => {
       const ma = multiaddr('/ip4/127.0.0.1/tcp/9090/ws/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
-
       const listener = ws.createListener((conn) => { })
 
-      listener.listen(ma, () => {
-        listener.getAddrs((err, addrs) => {
-          expect(err).to.not.exist()
+      await listener.listen(ma)
+      const addrs = await listener.getAddrs()
       expect(addrs.length).to.equal(1)
       expect(addrs[0]).to.deep.equal(ma)
-          listener.close(done)
-        })
-      })
+      await listener.close()
     })
   })
 
@@ -160,19 +137,18 @@ describe('listen', () => {
       ws = new WS()
     })
 
-    it('listen, check for callback', (done) => {
+    it('listen, check for promise', async () => {
       const listener = ws.createListener((conn) => { })
-
-      listener.listen(ma, () => {
-        listener.close(done)
-      })
+      await listener.listen(ma)
+      await listener.close()
     })
 
     it('listen, check for listening event', (done) => {
       const listener = ws.createListener((conn) => { })
 
-      listener.on('listening', () => {
-        listener.close(done)
+      listener.on('listening', async () => {
+        await listener.close()
+        done()
       })
 
       listener.listen(ma)
@@ -189,14 +165,11 @@ describe('listen', () => {
       listener.listen(ma)
     })
 
-    it('listen on addr with /ipfs/QmHASH', (done) => {
+    it('listen on addr with /ipfs/QmHASH', async () => {
       const ma = multiaddr('/ip6/::1/tcp/9091/ws/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
-
       const listener = ws.createListener((conn) => { })
-
-      listener.listen(ma, () => {
-        listener.close(done)
-      })
+      await listener.listen(ma)
+      await listener.close()
     })
   })
 })