Skip to content
This repository was archived by the owner on Feb 12, 2024. It is now read-only.

Commit 5078ddc

Browse files
daviddiasdaviddias
authored
Feature: PubSub 🌟 (#644)
feat: pubsub
1 parent 6a5afdd commit 5078ddc

File tree

24 files changed

+804
-151
lines changed

24 files changed

+804
-151
lines changed

‎package.json

+10-7
Original file line numberDiff line numberDiff line change
@@ -61,17 +61,18 @@
6161
"aegir": "^9.3.0",
6262
"buffer-loader": "0.0.1",
6363
"chai": "^3.5.0",
64+
"delay": "^1.3.1",
6465
"detect-node": "^2.0.3",
6566
"eslint-plugin-react": "^6.9.0",
6667
"execa": "^0.6.0",
6768
"expose-loader": "^0.7.1",
6869
"form-data": "^2.1.2",
6970
"fs-pull-blob-store": "^0.4.1",
7071
"gulp": "^3.9.1",
71-
"interface-ipfs-core": "^0.23.0",
72+
"interface-ipfs-core": "^0.23.3",
7273
"ipfsd-ctl": "^0.18.1",
7374
"left-pad": "^1.1.3",
74-
"lodash": "^4.17.2",
75+
"lodash": "^4.17.4",
7576
"mocha": "^3.2.0",
7677
"ncp": "^2.0.0",
7778
"nexpect": "^0.5.0",
@@ -85,11 +86,12 @@
8586
"async": "^2.1.4",
8687
"bl": "^1.2.0",
8788
"boom": "^4.2.0",
88-
"debug": "^2.5.1",
89+
"debug": "^2.6.0",
8990
"fs-pull-blob-store": "^0.3.0",
9091
"glob": "^7.1.1",
9192
"hapi": "^16.1.0",
9293
"hapi-set-header": "^1.0.2",
94+
"hoek": "^4.1.0",
9395
"idb-pull-blob-store": "^0.5.1",
9496
"ipfs-api": "^12.1.4",
9597
"ipfs-bitswap": "^0.9.0",
@@ -101,9 +103,10 @@
101103
"ipfs-unixfs-engine": "^0.15.0",
102104
"ipld-resolver": "^0.4.1",
103105
"isstream": "^0.1.2",
104-
"joi": "^10.0.6",
105-
"libp2p-ipfs-nodejs": "^0.17.1",
106-
"libp2p-ipfs-browser": "^0.17.3",
106+
"libp2p-floodsub": "0.7.1",
107+
"joi": "^10.1.0",
108+
"libp2p-ipfs-nodejs": "^0.17.3",
109+
"libp2p-ipfs-browser": "^0.17.4",
107110
"lodash.flatmap": "^4.5.0",
108111
"lodash.get": "^4.4.2",
109112
"lodash.has": "^4.5.2",
@@ -132,7 +135,7 @@
132135
"temp": "^0.8.3",
133136
"through2": "^2.0.3",
134137
"update-notifier": "^1.0.3",
135-
"yargs": "^6.5.0"
138+
"yargs": "^6.6.0"
136139
},
137140
"contributors": [
138141
"Andrew de Andrade <[email protected]>",

‎src/cli/commands/pubsub.js

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
'use strict'
2+
3+
module.exports = {
4+
command: 'pubsub',
5+
6+
description: 'pubsub commands',
7+
8+
builder (yargs) {
9+
return yargs
10+
.commandDir('pubsub')
11+
},
12+
13+
handler (argv) {}
14+
}

‎src/cli/commands/pubsub/ls.js

+32
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
'use strict'
2+
3+
const utils = require('../../utils')
4+
const debug = require('debug')
5+
const log = debug('cli:pubsub')
6+
log.error = debug('cli:pubsub:error')
7+
8+
module.exports = {
9+
command: 'ls',
10+
11+
describe: 'Get your list of subscriptions',
12+
13+
builder: {},
14+
15+
handler (argv) {
16+
utils.getIPFS((err, ipfs) => {
17+
if (err) {
18+
throw err
19+
}
20+
21+
ipfs.pubsub.ls((err, subscriptions) => {
22+
if (err) {
23+
throw err
24+
}
25+
26+
subscriptions.forEach((sub) => {
27+
console.log(sub)
28+
})
29+
})
30+
})
31+
}
32+
}

‎src/cli/commands/pubsub/peers.js

+32
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
'use strict'
2+
3+
const utils = require('../../utils')
4+
const debug = require('debug')
5+
const log = debug('cli:pubsub')
6+
log.error = debug('cli:pubsub:error')
7+
8+
module.exports = {
9+
command: 'peers <topic>',
10+
11+
describe: 'Get all peers subscribed to a topic',
12+
13+
builder: {},
14+
15+
handler (argv) {
16+
utils.getIPFS((err, ipfs) => {
17+
if (err) {
18+
throw err
19+
}
20+
21+
ipfs.pubsub.peers(argv.topic, (err, peers) => {
22+
if (err) {
23+
throw err
24+
}
25+
26+
peers.forEach((peer) => {
27+
console.log(peer)
28+
})
29+
})
30+
})
31+
}
32+
}

‎src/cli/commands/pubsub/pub.js

+30
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
'use strict'
2+
3+
const utils = require('../../utils')
4+
const debug = require('debug')
5+
const log = debug('cli:pubsub')
6+
log.error = debug('cli:pubsub:error')
7+
8+
module.exports = {
9+
command: 'pub <topic> <data>',
10+
11+
describe: 'Publish data to a topic',
12+
13+
builder: {},
14+
15+
handler (argv) {
16+
utils.getIPFS((err, ipfs) => {
17+
if (err) {
18+
throw err
19+
}
20+
21+
const data = new Buffer(String(argv.data))
22+
23+
ipfs.pubsub.publish(argv.topic, data, (err) => {
24+
if (err) {
25+
throw err
26+
}
27+
})
28+
})
29+
}
30+
}

‎src/cli/commands/pubsub/sub.js

+32
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
'use strict'
2+
3+
const utils = require('../../utils')
4+
const debug = require('debug')
5+
const log = debug('cli:pubsub')
6+
log.error = debug('cli:pubsub:error')
7+
8+
module.exports = {
9+
command: 'sub <topic>',
10+
11+
describe: 'Subscribe to a topic',
12+
13+
builder: {},
14+
15+
handler (argv) {
16+
utils.getIPFS((err, ipfs) => {
17+
if (err) {
18+
throw err
19+
}
20+
21+
const handler = (msg) => {
22+
console.log(msg.data.toString())
23+
}
24+
25+
ipfs.pubsub.subscribe(argv.topic, handler, (err) => {
26+
if (err) {
27+
throw err
28+
}
29+
})
30+
})
31+
}
32+
}

‎src/core/components/go-offline.js

+8-3
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
11
'use strict'
22

3-
module.exports = function goOffline (self) {
4-
return (cb) => {
3+
module.exports = (self) => {
4+
return (callback) => {
55
self._blockService.goOffline()
66
self._bitswap.stop()
7-
self.libp2p.stop(cb)
7+
self._pubsub.stop((err) => {
8+
if (err) {
9+
return callback(err)
10+
}
11+
self.libp2p.stop(callback)
12+
})
813
}
914
}

‎src/core/components/go-online.js

+22-8
Original file line numberDiff line numberDiff line change
@@ -2,25 +2,39 @@
22

33
const series = require('async/series')
44
const Bitswap = require('ipfs-bitswap')
5+
const FloodSub = require('libp2p-floodsub')
56

6-
module.exports = function goOnline (self) {
7-
return (cb) => {
7+
module.exports = (self) => {
8+
return (callback) => {
89
series([
9-
self.load,
10-
self.libp2p.start
10+
(cb) => self.load(cb),
11+
(cb) => self.libp2p.start(cb)
1112
], (err) => {
1213
if (err) {
13-
return cb(err)
14+
return callback(err)
1415
}
1516

1617
self._bitswap = new Bitswap(
1718
self._libp2pNode,
1819
self._repo.blockstore,
1920
self._libp2pNode.peerBook
2021
)
21-
self._bitswap.start()
22-
self._blockService.goOnline(self._bitswap)
23-
cb()
22+
23+
self._pubsub = new FloodSub(self._libp2pNode)
24+
25+
series([
26+
(cb) => {
27+
self._bitswap.start()
28+
cb()
29+
},
30+
(cb) => {
31+
self._blockService.goOnline(self._bitswap)
32+
cb()
33+
},
34+
(cb) => self._pubsub.start(cb) // ,
35+
// For all of the protocols to handshake with each other
36+
// (cb) => setTimeout(cb, 1000) // Still not decided if we want this
37+
], callback)
2438
})
2539
}
2640
}

‎src/core/components/pubsub.js

+97
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
'use strict'
2+
3+
const promisify = require('promisify-es6')
4+
const setImmediate = require('async/setImmediate')
5+
6+
const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
7+
8+
module.exports = function pubsub (self) {
9+
return {
10+
subscribe: (topic, options, handler, callback) => {
11+
if (!self.isOnline()) {
12+
throw OFFLINE_ERROR
13+
}
14+
15+
if (typeof options === 'function') {
16+
callback = handler
17+
handler = options
18+
options = {}
19+
}
20+
21+
if (!callback) {
22+
return new Promise((resolve, reject) => {
23+
subscribe(topic, options, handler, (err) => {
24+
if (err) {
25+
return reject(err)
26+
}
27+
resolve()
28+
})
29+
})
30+
}
31+
32+
subscribe(topic, options, handler, callback)
33+
},
34+
35+
unsubscribe: (topic, handler) => {
36+
const ps = self._pubsub
37+
38+
ps.removeListener(topic, handler)
39+
40+
if (ps.listenerCount(topic) === 0) {
41+
ps.unsubscribe(topic)
42+
}
43+
},
44+
45+
publish: promisify((topic, data, callback) => {
46+
if (!self.isOnline()) {
47+
return setImmediate(() => callback(OFFLINE_ERROR))
48+
}
49+
50+
if (!Buffer.isBuffer(data)) {
51+
return setImmediate(() => callback(new Error('data must be a Buffer')))
52+
}
53+
54+
self._pubsub.publish(topic, data)
55+
setImmediate(() => callback())
56+
}),
57+
58+
ls: promisify((callback) => {
59+
if (!self.isOnline()) {
60+
return setImmediate(() => callback(OFFLINE_ERROR))
61+
}
62+
63+
const subscriptions = Array.from(
64+
self._pubsub.subscriptions
65+
)
66+
67+
setImmediate(() => callback(null, subscriptions))
68+
}),
69+
70+
peers: promisify((topic, callback) => {
71+
if (!self.isOnline()) {
72+
return setImmediate(() => callback(OFFLINE_ERROR))
73+
}
74+
75+
const peers = Array.from(self._pubsub.peers.values())
76+
.filter((peer) => peer.topics.has(topic))
77+
.map((peer) => peer.info.id.toB58String())
78+
79+
setImmediate(() => callback(null, peers))
80+
}),
81+
82+
setMaxListeners (n) {
83+
return self._pubsub.setMaxListeners(n)
84+
}
85+
}
86+
87+
function subscribe (topic, options, handler, callback) {
88+
const ps = self._pubsub
89+
90+
if (ps.listenerCount(topic) === 0) {
91+
ps.subscribe(topic)
92+
}
93+
94+
ps.on(topic, handler)
95+
setImmediate(() => callback())
96+
}
97+
}

‎src/core/index.js

+3
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ const swarm = require('./components/swarm')
2323
const ping = require('./components/ping')
2424
const files = require('./components/files')
2525
const bitswap = require('./components/bitswap')
26+
const pubsub = require('./components/pubsub')
2627

2728
exports = module.exports = IPFS
2829

@@ -44,6 +45,7 @@ function IPFS (repoInstance) {
4445
this._bitswap = null
4546
this._blockService = new BlockService(this._repo)
4647
this._ipldResolver = new IPLDResolver(this._blockService)
48+
this._pubsub = null
4749

4850
// IPFS Core exposed components
4951

@@ -67,4 +69,5 @@ function IPFS (repoInstance) {
6769
this.files = files(this)
6870
this.bitswap = bitswap(this)
6971
this.ping = ping(this)
72+
this.pubsub = pubsub(this)
7073
}

0 commit comments

Comments
 (0)