Skip to content

Commit

Permalink
fix: promisified methods (#398)
Browse files Browse the repository at this point in the history
* chore: update ws rendezvous dep

test(fix): fix tests with latest ws rendezvous server

* fix: promisification of libp2p methods

test: add tests to verify promisify support until async/await endeavor is complete

* chore: fix linting
  • Loading branch information
jacobheun authored Aug 6, 2019
1 parent 9a8d609 commit ff7a6c8
Show file tree
Hide file tree
Showing 9 changed files with 169 additions and 108 deletions.
93 changes: 35 additions & 58 deletions .aegir.js
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
'use strict'

const pull = require('pull-stream')
const parallel = require('async/parallel')
const WebSocketStarRendezvous = require('libp2p-websocket-star-rendezvous')
const sigServer = require('libp2p-webrtc-star/src/sig-server')

Expand All @@ -15,68 +14,46 @@ const {
let wrtcRendezvous
let wsRendezvous
let node
let peerInfo

const before = (done) => {
parallel([
(cb) => {
sigServer.start({
port: WRTC_RENDEZVOUS_MULTIADDR.nodeAddress().port
// cryptoChallenge: true TODO: needs https://github.com/libp2p/js-libp2p-webrtc-star/issues/128
})
.then(server => {
wrtcRendezvous = server
cb()
})
.catch(cb)
},
(cb) => {
WebSocketStarRendezvous.start({
port: WS_RENDEZVOUS_MULTIADDR.nodeAddress().port,
refreshPeerListIntervalMS: 1000,
strictMultiaddr: false,
cryptoChallenge: true
}, (err, _server) => {
if (err) {
return cb(err)
}
wsRendezvous = _server
cb()
})
},
(cb) => {
getPeerRelay((err, peerInfo) => {
if (err) {
return done(err)
}

node = new Node({
peerInfo,
config: {
relay: {
enabled: true,
hop: {
enabled: true,
active: true
}
}
}
})
const before = async () => {
[wrtcRendezvous, wsRendezvous, peerInfo] = await Promise.all([
sigServer.start({
port: WRTC_RENDEZVOUS_MULTIADDR.nodeAddress().port
// cryptoChallenge: true TODO: needs https://github.com/libp2p/js-libp2p-webrtc-star/issues/128
}),
WebSocketStarRendezvous.start({
port: WS_RENDEZVOUS_MULTIADDR.nodeAddress().port,
refreshPeerListIntervalMS: 1000,
strictMultiaddr: false,
cryptoChallenge: true
}),
getPeerRelay()
])

node.handle('/echo/1.0.0', (protocol, conn) => pull(conn, conn))
node.start(cb)
})
node = new Node({
peerInfo,
config: {
relay: {
enabled: true,
hop: {
enabled: true,
active: true
}
}
}
], done)
})

node.handle('/echo/1.0.0', (protocol, conn) => pull(conn, conn))
await node.start()
}

const after = (done) => {
setTimeout(() =>
parallel([
(cb) => wrtcRendezvous.stop().then(cb).catch(cb),
...[node, wsRendezvous].map((s) => (cb) => s.stop(cb)),
], done),
2000
)
const after = () => {
return Promise.all([
wrtcRendezvous.stop(),
wsRendezvous.stop(),
node.stop()
])
}

module.exports = {
Expand Down
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@
"libp2p-tcp": "^0.13.0",
"libp2p-webrtc-star": "^0.16.1",
"libp2p-websocket-star": "~0.10.2",
"libp2p-websocket-star-rendezvous": "~0.3.0",
"libp2p-websocket-star-rendezvous": "~0.4.1",
"lodash.times": "^4.3.2",
"merge-options": "^1.0.1",
"nock": "^10.0.6",
Expand Down
24 changes: 19 additions & 5 deletions src/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,11 @@ class Libp2p extends EventEmitter {
})

this._peerDiscovered = this._peerDiscovered.bind(this)

// promisify all instance methods
;['start', 'stop', 'dial', 'dialProtocol', 'dialFSM', 'hangUp', 'ping'].forEach(method => {
this[method] = promisify(this[method], { context: this })
})
}

/**
Expand Down Expand Up @@ -295,6 +300,13 @@ class Libp2p extends EventEmitter {
})
}

/**
* Disconnects from the given peer
*
* @param {PeerInfo|PeerId|Multiaddr|string} peer The peer to ping
* @param {function(Error)} callback
* @returns {void}
*/
hangUp (peer, callback) {
this._getPeerInfo(peer, (err, peerInfo) => {
if (err) { return callback(err) }
Expand All @@ -303,6 +315,13 @@ class Libp2p extends EventEmitter {
})
}

/**
* Pings the provided peer
*
* @param {PeerInfo|PeerId|Multiaddr|string} peer The peer to ping
* @param {function(Error, Ping)} callback
* @returns {void}
*/
ping (peer, callback) {
if (!this.isStarted()) {
return callback(notStarted('ping', this.state._state))
Expand Down Expand Up @@ -550,11 +569,6 @@ class Libp2p extends EventEmitter {
}
}

// promisify all instance methods
['start', 'stop', 'dial', 'dialProtocol', 'dialFSM', 'hangUp', 'ping'].forEach(method => {
Libp2p[method] = promisify(Libp2p[method])
})

module.exports = Libp2p
/**
* Like `new Libp2p(options)` except it will create a `PeerInfo`
Expand Down
7 changes: 1 addition & 6 deletions test/circuit-relay.browser.js
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,7 @@ describe('circuit relay', () => {
let peerRelay

before('get peer relay', async () => {
peerRelay = await new Promise(resolve => {
getPeerRelay((err, peer) => {
expect(err).to.not.exist()
resolve(peer)
})
})
peerRelay = await getPeerRelay()
})

before('create the browser nodes', async () => {
Expand Down
1 change: 1 addition & 0 deletions test/node.js
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ require('./stream-muxing.node')
require('./peer-discovery.node')
require('./peer-routing.node')
require('./ping.node')
require('./promisify.node')
require('./pubsub.node')
require('./content-routing.node')
require('./circuit-relay.node')
Expand Down
87 changes: 87 additions & 0 deletions test/promisify.node.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
/* eslint-env mocha */
'use strict'

/**
* This test suite is intended to validate compatability of
* the promisified api, until libp2p has been fully migrated to
* async/await. Once the migration is complete and all tests
* are using async/await, this file can be removed.
*/

const chai = require('chai')
chai.use(require('dirty-chai'))
const expect = chai.expect
const promisify = require('promisify-es6')
const createNode = promisify(require('./utils/create-node'))
const { createPeerInfo } = require('./utils/create-node')
const Node = require('./utils/bundle-nodejs')
const pull = require('pull-stream')
const Ping = require('libp2p-ping')

/**
* As libp2p is currently promisified, when extending libp2p,
* method arguments must be passed to `super` to ensure the
* promisify callbacks are properly resolved
*/
class AsyncLibp2p extends Node {
async start (...args) {
await super.start(...args)
}

async stop (...args) {
await super.start(...args)
}
}

async function createAsyncNode () {
const peerInfo = await promisify(createPeerInfo)()
peerInfo.multiaddrs.add('/ip4/0.0.0.0/tcp/0')
return new AsyncLibp2p({ peerInfo })
}

describe('promisified libp2p', () => {
let libp2p
let otherNode
const ECHO_PROTO = '/echo/1.0.0'

before('Create and Start', async () => {
[libp2p, otherNode] = await Promise.all([
createNode('/ip4/0.0.0.0/tcp/0'),
createAsyncNode()
])

return [libp2p, otherNode].map(node => {
node.handle(ECHO_PROTO, (_, conn) => pull(conn, conn))
return node.start()
})
})

after('Stop', () => {
return [libp2p, otherNode].map(node => node.stop())
})

afterEach('Hang up', () => {
return libp2p.hangUp(otherNode.peerInfo)
})

it('dial', async () => {
const stream = await libp2p.dial(otherNode.peerInfo)
expect(stream).to.not.exist()
expect(libp2p._switch.connection.getAll()).to.have.length(1)
})

it('dialFSM', async () => {
const connectionFSM = await libp2p.dialFSM(otherNode.peerInfo, ECHO_PROTO)
expect(connectionFSM).to.exist()
})

it('dialProtocol', async () => {
const stream = await libp2p.dialProtocol(otherNode.peerInfo, ECHO_PROTO)
expect(stream).to.exist()
})

it('ping', async () => {
const ping = await libp2p.ping(otherNode.peerInfo)
expect(ping).to.be.an.instanceOf(Ping)
})
})
13 changes: 5 additions & 8 deletions test/transports.browser.js
Original file line number Diff line number Diff line change
Expand Up @@ -25,14 +25,11 @@ describe('transports', () => {
let peerBMultiaddr
let nodeA

before((done) => {
getPeerRelay((err, peerInfo) => {
expect(err).to.not.exist()
peerB = new PeerInfo(peerInfo.id)
peerBMultiaddr = `/ip4/127.0.0.1/tcp/9200/ws/p2p/${peerInfo.id.toB58String()}`
peerB.multiaddrs.add(peerBMultiaddr)
done()
})
before(async () => {
const peerInfo = await getPeerRelay()
peerB = new PeerInfo(peerInfo.id)
peerBMultiaddr = `/ip4/127.0.0.1/tcp/9200/ws/p2p/${peerInfo.id.toB58String()}`
peerB.multiaddrs.add(peerBMultiaddr)
})

after((done) => nodeA.stop(done))
Expand Down
41 changes: 15 additions & 26 deletions test/transports.node.js
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@ const createNode = require('./utils/create-node.js')
const tryEcho = require('./utils/try-echo')
const echo = require('./utils/echo')

const { WRTC_RENDEZVOUS_MULTIADDR } = require('./utils/constants')
const {
WRTC_RENDEZVOUS_MULTIADDR
} = require('./utils/constants')

describe('transports', () => {
describe('TCP only', () => {
Expand Down Expand Up @@ -576,25 +578,24 @@ describe('transports', () => {
let nodeTCP
let nodeWS
let nodeWebSocketStar

let ss
const PORT = 24642

before(async () => {
ss = await rendezvous.start({
port: PORT
})
})

before((done) => {
parallel([
(cb) => {
rendezvous.start({ port: 24642 }, (err, server) => {
expect(err).to.not.exist()
ss = server
cb()
})
},
(cb) => {
const wstar = new WSStar()

createNode([
'/ip4/0.0.0.0/tcp/0',
'/ip4/127.0.0.1/tcp/25011/ws',
'/ip4/127.0.0.1/tcp/24642/ws/p2p-websocket-star'
`/ip4/127.0.0.1/tcp/${PORT}/ws/p2p-websocket-star`
], {
modules: {
transport: [
Expand All @@ -603,13 +604,6 @@ describe('transports', () => {
wstar
],
peerDiscovery: [wstar.discovery]
},
config: {
peerDiscovery: {
[wstar.discovery.tag]: {
enabled: true
}
}
}
}, (err, node) => {
expect(err).to.not.exist()
Expand Down Expand Up @@ -640,18 +634,11 @@ describe('transports', () => {
const wstar = new WSStar({})

createNode([
'/ip4/127.0.0.1/tcp/24642/ws/p2p-websocket-star'
`/ip4/127.0.0.1/tcp/${PORT}/ws/p2p-websocket-star`
], {
modules: {
transport: [wstar],
peerDiscovery: [wstar.discovery]
},
config: {
peerDiscovery: {
[wstar.discovery.tag]: {
enabled: true
}
}
}
}, (err, node) => {
expect(err).to.not.exist()
Expand All @@ -670,7 +657,9 @@ describe('transports', () => {
(cb) => nodeTCP.stop(cb),
(cb) => nodeWS.stop(cb),
(cb) => nodeWebSocketStar.stop(cb),
(cb) => ss.stop(cb)
async () => {
await ss.stop()
}
], done)
})

Expand Down
Loading

0 comments on commit ff7a6c8

Please sign in to comment.