Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Tutorial: Publish Subscribe #103

Merged
merged 1 commit into from
Jul 21, 2017
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions examples/pubsub/1.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
'use strict'

const libp2p = require('libp2p')
const TCP = require('libp2p-tcp')
const Multiplex = require('libp2p-multiplex')
const SECIO = require('libp2p-secio')
const PeerInfo = require('peer-info')
const MulticastDNS = require('libp2p-mdns')
const FloodSub = require('libp2p-floodsub')
const waterfall = require('async/waterfall')
const parallel = require('async/parallel')
const series = require('async/series')

class MyBundle extends libp2p {
constructor (peerInfo) {
const modules = {
transport: [new TCP()],
connection: {
muxer: [Multiplex],
crypto: [SECIO]
},
discovery: [new MulticastDNS(peerInfo, { interval: 2000 })]
}
super(modules, peerInfo)
}
}

function createNode (callback) {
let node

waterfall([
(cb) => PeerInfo.create(cb),
(peerInfo, cb) => {
peerInfo.multiaddrs.add('/ip4/0.0.0.0/tcp/0')
node = new MyBundle(peerInfo)
node.start(cb)
}
], (err) => callback(err, node))
}

parallel([
(cb) => createNode(cb),
(cb) => createNode(cb)
], (err, nodes) => {
if (err) { throw err }

const node1 = nodes[0]
const node2 = nodes[1]

const fs1 = new FloodSub(node1)
const fs2 = new FloodSub(node2)

series([
(cb) => fs1.start(cb),
(cb) => fs2.start(cb),
(cb) => node1.once('peer:discovery', (peer) => node1.dial(peer, cb)),
(cb) => setTimeout(cb, 500)
], (err) => {
if (err) { throw err }

fs2.on('news', (msg) => console.log(msg.from, msg.data.toString()))
fs2.subscribe('news')

setInterval(() => {
fs1.publish('news', Buffer.from('Bird bird bird, bird is the word!'))
}, 1000)
})
})
62 changes: 60 additions & 2 deletions examples/pubsub/README.md
Original file line number Diff line number Diff line change
@@ -1,2 +1,60 @@
# WIP - This example is still in the works
![](http://1.bp.blogspot.com/-tNvSnCW0KlQ/U-KOKGVoJkI/AAAAAAAAA3Q/aiSLMeSJFtw/s1600/WIP-sign.jpg)
# Publish Subscribe

Publish Subscribe is something out of scope for the modular networking stack that is libp2p, however, it is something that is enabled through the primitives that libp2p offers and so it has become one of the most interesting use cases for libp2p.

Currently, we have a PubSub implementation, [libp2p-floodsub](https://github.com/libp2p/js-libp2p-floodsub) and many more being researched at [research-pubsub](https://github.com/libp2p/research-pubsub).

We've seen many interesting use cases appear with this, here are some highlights:

- [Collaborative Text Editing](https://www.youtube.com/watch?v=-kdx8rJd8rQ)
- [IPFS PubSub (using libp2p-floodsub) for IoT](https://www.youtube.com/watch?v=qLpM5pBDGiE).
- [Real Time distributed Applications](https://www.youtube.com/watch?v=vQrbxyDPSXg)

## 1. Setting up a simple PubSub network on top of libp2p

For this example, we will use MulticastDNS for automatic Peer Discovery and libp2p-floodsub to give us the PubSub primitives we are looking for. This example is based the previous examples found in [Discovery Mechanisms](../discovery-mechanisms). You can find the complete version at [1.js](./1.js).

Getting PubSub is super simple, all you have to do is require the FloodSub module and pass it in a libp2p node, once you have done that you can start subscribing and publishing in any topic.

```JavaScript
const FloodSub = require('libp2p-floodsub')

const fs1 = new FloodSub(node1)
const fs2 = new FloodSub(node2)

series([
(cb) => fs1.start(cb),
(cb) => fs2.start(cb),
(cb) => node1.once('peer:discovery', (peer) => node1.dial(peer, cb)),
(cb) => setTimeout(cb, 500)
], (err) => {
if (err) { throw err }

fs2.on('news', (msg) => console.log(msg.from, msg.data.toString()))
fs2.subscribe('news')

setInterval(() => {
fs1.publish('news', Buffer.from('Bird bird bird, bird is the word!'))
}, 1000)
})
```

The output of the program should look like:

```
> node 1.js
QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82 Bird bird bird, bird is the word!
QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82 Bird bird bird, bird is the word!
QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82 Bird bird bird, bird is the word!
QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82 Bird bird bird, bird is the word!
QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82 Bird bird bird, bird is the word!
```

## 2. Future work

libp2p/IPFS PubSub is enabling a whole set of Distributed Real Time applications using CRDT (Conflict-Free Replicated Data Types). It is still going through heavy research (and hacking) and we invite you to join the conversation at [research-CRDT](https://github.com/ipfs/research-CRDT). Here is a list of some of the exciting examples:

- [PubSub Room](https://github.com/ipfs-labs/ipfs-pubsub-room)
- [Live DB - A always in Sync DB using CRDT](https://github.com/ipfs-labs/ipfs-live-db)
- [IIIF Annotations over IPFS, CRDT and libp2p](https://www.youtube.com/watch?v=hmAniA6g9D0&feature=youtu.be&t=10m40s)
- [orbit.chat - p2p chat application, fully running in the browser with js-ipfs, js-libp2p and orbit-db](http://orbit.chat/)