callbacks -> async / await

package.json

@@ -38,11 +38,15 @@
     "dirty-chai": "^2.0.1"
   },
   "dependencies": {
+    "abort-controller": "^3.0.0",
+    "async-iterator-to-pull-stream": "^1.3.0",
     "chai": "^4.2.0",
+    "interface-connection": "~0.3.3",
+    "it-goodbye": "^2.0.0",
+    "it-pipe": "^1.0.0",
     "multiaddr": "^5.0.2",
-    "pull-goodbye": "~0.0.2",
-    "pull-serializer": "~0.3.2",
-    "pull-stream": "^3.6.9"
+    "pull-stream": "^3.6.9",
+    "streaming-iterables": "^4.0.2"
   },
   "contributors": [
     "David Dias <daviddias.p@gmail.com>",

src/adapter.js

@@ -0,0 +1,80 @@
+'use strict'
+
+const { Connection } = require('interface-connection')
+const toPull = require('async-iterator-to-pull-stream')
+const error = require('pull-stream/sources/error')
+const drain = require('pull-stream/sinks/drain')
+const noop = () => {}
+
+function callbackify (fn) {
+  return async function (...args) {
+    let cb = args.pop()
+    if (typeof cb !== 'function') {
+      args.push(cb)
+      cb = noop
+    }
+    let res
+    try {
+      res = await fn(...args)
+    } catch (err) {
+      return cb(err)
+    }
+    cb(null, res)
+  }
+}
+
+// Legacy adapter to old transport & connection interface
+class Adapter {
+  constructor (transport) {
+    this.transport = transport
+  }
+
+  dial (ma, options, callback) {
+    if (typeof options === 'function') {
+      callback = options
+      options = {}
+    }
+
+    callback = callback || noop
+
+    const conn = new Connection()
+
+    this.transport.dial(ma, options)
+      .then(socket => {
+        conn.setInnerConn(toPull.duplex(socket))
+        conn.getObservedAddrs = callbackify(socket.getObservedAddrs.bind(socket))
+        conn.close = callbackify(socket.close.bind(socket))
+        callback(null, conn)
+      })
+      .catch(err => {
+        conn.setInnerConn({ sink: drain(), source: error(err) })
+        callback(err)
+      })
+
+    return conn
+  }
+
+  createListener (options, handler) {
+    if (typeof options === 'function') {
+      handler = options
+      options = {}
+    }
+
+    const server = this.transport.createListener(options, socket => {
+      const conn = new Connection(toPull.duplex(socket))
+      conn.getObservedAddrs = callbackify(socket.getObservedAddrs.bind(socket))
+      handler(conn)
+    })
+
+    const proxy = {
+      listen: callbackify(server.listen.bind(server)),
+      close: callbackify(server.close.bind(server)),
+      getAddrs: callbackify(server.getAddrs.bind(server)),
+      getObservedAddrs: callbackify(() => server.getObservedAddrs())
+    }
+
+    return new Proxy(server, { get: (_, prop) => proxy[prop] || server[prop] })
+  }
+}
+
+module.exports = Adapter

src/dial-test.js

@@ -5,69 +5,86 @@ const chai = require('chai')
 const dirtyChai = require('dirty-chai')
 const expect = chai.expect
 chai.use(dirtyChai)
-const pull = require('pull-stream')
-const goodbye = require('pull-goodbye')
-const serializer = require('pull-serializer')
+
+const goodbye = require('it-goodbye')
+const { collect } = require('streaming-iterables')
+const pipe = require('it-pipe')
+const AbortController = require('abort-controller')
+const AbortError = require('./errors').AbortError
 
 module.exports = (common) => {
   describe('dial', () => {
     let addrs
     let transport
+    let connector
     let listener
 
-    before((done) => {
-      common.setup((err, _transport, _addrs) => {
-        if (err) return done(err)
-        transport = _transport
-        addrs = _addrs
-        done()
-      })
+    before(async () => {
+      ({ addrs, transport, connector } = await common.setup())
     })
 
-    after((done) => {
-      common.teardown(done)
+    after(() => common.teardown && common.teardown())
+
+    beforeEach(() => {
+      listener = transport.createListener((conn) => pipe(conn, conn))
+      return listener.listen(addrs[0])
     })
 
-    beforeEach((done) => {
-      listener = transport.createListener((conn) => {
-        pull(conn, conn)
-      })
-      listener.listen(addrs[0], done)
+    afterEach(() => listener.close())
+
+    it('simple', async () => {
+      const conn = await transport.dial(addrs[0])
+
+      const s = goodbye({ source: ['hey'], sink: collect })
+
+      const result = await pipe(s, conn, s)
+
+      expect(result.length).to.equal(1)
+      expect(result[0].toString()).to.equal('hey')
     })
 
-    afterEach((done) => {
-      listener.close(done)
+    it('to non existent listener', async () => {
+      try {
+        await transport.dial(addrs[1])
+      } catch (_) {
+        // Success: expected an error to be throw
+        return
+      }
+      expect.fail('Did not throw error attempting to connect to non-existent listener')
     })
 
-    it('simple', (done) => {
-      const s = serializer(goodbye({
-        source: pull.values(['hey']),
-        sink: pull.collect((err, values) => {
-          expect(err).to.not.exist()
-          expect(
-            values
-          ).to.be.eql(
-            ['hey']
-          )
-          done()
-        })
-      }))
-
-      pull(
-        s,
-        transport.dial(addrs[0]),
-        s
-      )
+    it('cancel before dialing', async () => {
+      const controller = new AbortController()
+      controller.abort()
+      const socket = transport.dial(addrs[0], { signal: controller.signal })
+
+      try {
+        await socket
+      } catch (err) {
+        expect(err.code).to.eql(AbortError.code)
+        return
+      }
+      expect.fail('Did not throw error with code ' + AbortError.code)
     })
 
-    it('to non existent listener', (done) => {
-      pull(
-        transport.dial(addrs[1]),
-        pull.onEnd((err) => {
-          expect(err).to.exist()
-          done()
-        })
-      )
+    it('cancel while dialing', async () => {
+      // Add a delay to connect() so that we can cancel while the dial is in
+      // progress
+      connector.delay(100)
+
+      const controller = new AbortController()
+      const socket = transport.dial(addrs[0], { signal: controller.signal })
+      setTimeout(() => controller.abort(), 50)
+
+      try {
+        await socket
+      } catch (err) {
+        expect(err.code).to.eql(AbortError.code)
+        return
+      } finally {
+        connector.restore()
+      }
+      expect.fail('Did not throw error with code ' + AbortError.code)
     })
   })
 }

src/errors.js

@@ -0,0 +1,16 @@
+'use strict'
+
+class AbortError extends Error {
+  constructor () {
+    super('AbortError')
+    this.code = AbortError.code
+  }
+
+  static get code () {
+    return 'ABORT_ERR'
+  }
+}
+
+module.exports = {
+  AbortError
+}

src/index.js

@@ -10,3 +10,6 @@ module.exports = (common) => {
     listen(common)
   })
 }
+
+module.exports.AbortError = require('./errors').AbortError
+module.exports.Adapter = require('./adapter')