diff --git a/README.md b/README.md index 5d853e00c..7f29a44b4 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,30 @@ $ ipfs config --json API.HTTPHeaders.Access-Control-Allow-Methods "[\"PUT\", \"P > `js-ipfs-api` follows the spec defined by [`interface-ipfs-core`](https://github.com/ipfs/interface-ipfs-core), which concerns the interface to expect from IPFS implementations. This interface is a currently active endeavor. You can use it today to consult the methods available. +#### Caveats + +##### Pubsub + +**Currently, the [PubSub API only works in Node.js envinroment](https://github.com/ipfs/js-ipfs-api/issues/518)** + +We currently don't support pubsub when run in the browser, and we test it with separate set of tests to make sure if it's being used in the browser, pubsub errors. + +More info: https://github.com/ipfs/js-ipfs-api/issues/518 + +This means: +- You can use pubsub from js-ipfs-api in Node.js +- You can use pubsub from js-ipfs-api in Electron + (when js-ipfs-api is ran in the main process of Electron) +- You can't use pubsub from js-ipfs-api in the browser +- You can't use pubsub from js-ipfs-api in Electron's + renderer process +- You can use pubsub from js-ipfs in the browsers +- You can use pubsub from js-ipfs in Node.js +- You can use pubsub from js-ipfs in Electron + (in both the main process and the renderer process) +- See https://github.com/ipfs/js-ipfs for details on + pubsub in js-ipfs + ##### [bitswap]() - [`ipfs.bitswap.wantlist()`]() diff --git a/package.json b/package.json index a8f610197..eede6ebe4 100644 --- a/package.json +++ b/package.json @@ -66,7 +66,7 @@ "eslint-plugin-react": "^6.10.3", "gulp": "^3.9.1", "hapi": "^16.1.0", - "interface-ipfs-core": "~0.26.1", + "interface-ipfs-core": "~0.26.2", "ipfsd-ctl": "~0.20.0", "pre-commit": "^1.2.2", "socket.io": "^1.7.3", diff --git a/src/api/pubsub.js b/src/api/pubsub.js new file mode 100644 index 000000000..1afd05fab --- /dev/null +++ b/src/api/pubsub.js @@ -0,0 +1,162 @@ +'use strict' + +const promisify = require('promisify-es6') +const EventEmitter = require('events') +const eos = require('end-of-stream') +const isNode = require('detect-node') +const PubsubMessageStream = require('../pubsub-message-stream') +const stringlistToArray = require('../stringlist-to-array') + +const NotSupportedError = () => new Error('pubsub is currently not supported when run in the browser') + +/* Public API */ +module.exports = (send) => { + /* Internal subscriptions state and functions */ + const ps = new EventEmitter() + const subscriptions = {} + ps.id = Math.random() + return { + subscribe: (topic, options, handler, callback) => { + const defaultOptions = { + discover: false + } + + if (typeof options === 'function') { + callback = handler + handler = options + options = defaultOptions + } + + if (!options) { + options = defaultOptions + } + + // Throw an error if ran in the browsers + if (!isNode) { + if (!callback) { + return Promise.reject(NotSupportedError()) + } + return callback(NotSupportedError()) + } + + // promisify doesn't work as we always pass a + // function as last argument (`handler`) + if (!callback) { + return new Promise((resolve, reject) => { + subscribe(topic, options, handler, (err) => { + if (err) { + return reject(err) + } + resolve() + }) + }) + } + + subscribe(topic, options, handler, callback) + }, + unsubscribe: (topic, handler) => { + if (!isNode) { + throw NotSupportedError() + } + + if (ps.listenerCount(topic) === 0 || !subscriptions[topic]) { + throw new Error(`Not subscribed to '${topic}'`) + } + + ps.removeListener(topic, handler) + + // Drop the request once we are actualy done + if (ps.listenerCount(topic) === 0) { + subscriptions[topic].abort() + subscriptions[topic] = null + } + }, + publish: promisify((topic, data, callback) => { + if (!isNode) { + return callback(NotSupportedError()) + } + + if (!Buffer.isBuffer(data)) { + return callback(new Error('data must be a Buffer')) + } + + const request = { + path: 'pubsub/pub', + args: [topic, data] + } + + send(request, callback) + }), + ls: promisify((callback) => { + if (!isNode) { + return callback(NotSupportedError()) + } + + const request = { + path: 'pubsub/ls' + } + + send.andTransform(request, stringlistToArray, callback) + }), + peers: promisify((topic, callback) => { + if (!isNode) { + return callback(NotSupportedError()) + } + + const request = { + path: 'pubsub/peers', + args: [topic] + } + + send.andTransform(request, stringlistToArray, callback) + }), + setMaxListeners (n) { + return ps.setMaxListeners(n) + } + } + + function subscribe (topic, options, handler, callback) { + ps.on(topic, handler) + if (subscriptions[topic]) { + return callback() + } + + // Request params + const request = { + path: 'pubsub/sub', + args: [topic], + qs: { + discover: options.discover + } + } + + // Start the request and transform the response + // stream to Pubsub messages stream + subscriptions[topic] = send.andTransform(request, PubsubMessageStream.from, (err, stream) => { + if (err) { + subscriptions[topic] = null + ps.removeListener(topic, handler) + return callback(err) + } + + stream.on('data', (msg) => { + ps.emit(topic, msg) + }) + + stream.on('error', (err) => { + ps.emit('error', err) + }) + + eos(stream, (err) => { + if (err) { + ps.emit('error', err) + } + + subscriptions[topic] = null + ps.removeListener(topic, handler) + }) + + callback() + }) + } +} diff --git a/src/load-commands.js b/src/load-commands.js index b69c197cb..1246ef5ef 100644 --- a/src/load-commands.js +++ b/src/load-commands.js @@ -25,6 +25,7 @@ function requireCommands () { refs: require('./api/refs'), repo: require('./api/repo'), swarm: require('./api/swarm'), + pubsub: require('./api/pubsub'), update: require('./api/update'), version: require('./api/version') } diff --git a/src/pubsub-message-stream.js b/src/pubsub-message-stream.js new file mode 100644 index 000000000..f994cc401 --- /dev/null +++ b/src/pubsub-message-stream.js @@ -0,0 +1,33 @@ +'use strict' + +const TransformStream = require('readable-stream').Transform +const PubsubMessage = require('./pubsub-message-utils') + +class PubsubMessageStream extends TransformStream { + constructor (options) { + const opts = Object.assign(options || {}, { objectMode: true }) + super(opts) + } + + static from (inputStream, callback) { + let outputStream = inputStream.pipe(new PubsubMessageStream()) + inputStream.on('end', () => outputStream.emit('end')) + callback(null, outputStream) + } + + _transform (obj, enc, callback) { + let msg + try { + msg = PubsubMessage.deserialize(obj, 'base64') + } catch (err) { + // Not a valid pubsub message + // go-ipfs returns '{}' as the very first object atm, we skip that + return callback() + } + + this.push(msg) + callback() + } +} + +module.exports = PubsubMessageStream diff --git a/src/pubsub-message-utils.js b/src/pubsub-message-utils.js new file mode 100644 index 000000000..ae7ec6bab --- /dev/null +++ b/src/pubsub-message-utils.js @@ -0,0 +1,39 @@ +'use strict' + +const bs58 = require('bs58') + +module.exports = { + deserialize (data, enc) { + enc = enc ? enc.toLowerCase() : 'json' + + if (enc === 'json') { + return deserializeFromJson(data) + } else if (enc === 'base64') { + return deserializeFromBase64(data) + } + + throw new Error(`Unsupported encoding: '${enc}'`) + } +} + +function deserializeFromJson (data) { + const json = JSON.parse(data) + return deserializeFromBase64(json) +} + +function deserializeFromBase64 (obj) { + if (!isPubsubMessage(obj)) { + throw new Error(`Not a pubsub message`) + } + + return { + from: bs58.encode(new Buffer(obj.from, 'base64')).toString(), + seqno: new Buffer(obj.seqno, 'base64'), + data: new Buffer(obj.data, 'base64'), + topicCIDs: obj.topicIDs || obj.topicCIDs + } +} + +function isPubsubMessage (obj) { + return obj && obj.from && obj.seqno && obj.data && (obj.topicIDs || obj.topicCIDs) +} diff --git a/test/interface/pubsub.spec.js b/test/interface/pubsub.spec.js new file mode 100644 index 000000000..5448ccc97 --- /dev/null +++ b/test/interface/pubsub.spec.js @@ -0,0 +1,23 @@ +/* eslint-env mocha */ + +'use strict' + +const test = require('interface-ipfs-core') +const FactoryClient = require('../ipfs-factory/client') +const isNode = require('detect-node') + +if (isNode) { + let fc + + const common = { + setup: function (callback) { + fc = new FactoryClient() + callback(null, fc) + }, + teardown: function (callback) { + fc.dismantle(callback) + } + } + + test.pubsub(common) +} diff --git a/test/ipfs-factory/daemon-spawner.js b/test/ipfs-factory/daemon-spawner.js index 7904ea43e..0d861c402 100644 --- a/test/ipfs-factory/daemon-spawner.js +++ b/test/ipfs-factory/daemon-spawner.js @@ -84,7 +84,7 @@ function spawnEphemeralNode (callback) { node.setConfig(configKey, configVal, cb) }, cb) }, - (cb) => node.startDaemon(cb) + (cb) => node.startDaemon(['--enable-pubsub-experiment'], cb) ], (err) => callback(err, node)) }) } diff --git a/test/pubsub-in-browser.spec.js b/test/pubsub-in-browser.spec.js new file mode 100644 index 000000000..53b50d1ec --- /dev/null +++ b/test/pubsub-in-browser.spec.js @@ -0,0 +1,182 @@ +/* + We currently don't support pubsub when run in the browser, + and we test it with separate set of tests to make sure + if it's being used in the browser, pubsub errors. + + More info: https://github.com/ipfs/js-ipfs-api/issues/518 + + This means: + - You can use pubsub from js-ipfs-api in Node.js + - You can use pubsub from js-ipfs-api in Electron + (when js-ipfs-api is ran in the main process of Electron) + + - You can't use pubsub from js-ipfs-api in the browser + - You can't use pubsub from js-ipfs-api in Electron's + renderer process + + - You can use pubsub from js-ipfs in the browsers + - You can use pubsub from js-ipfs in Node.js + - You can use pubsub from js-ipfs in Electron + (in both the main process and the renderer process) + - See https://github.com/ipfs/js-ipfs for details on + pubsub in js-ipfs +*/ + +/* eslint-env mocha */ +/* eslint max-nested-callbacks: ['error', 8] */ +'use strict' + +const series = require('async/series') +const waterfall = require('async/waterfall') +const isNode = require('detect-node') +const FactoryClient = require('./ipfs-factory/client') +const chai = require('chai') +const dirtyChai = require('dirty-chai') +const expect = chai.expect +chai.use(dirtyChai) + +const expectedError = 'pubsub is currently not supported when run in the browser' + +function spawnWithId (factory, callback) { + waterfall([ + (cb) => factory.spawnNode(cb), + (node, cb) => node.id((err, res) => { + if (err) { + return cb(err) + } + node.peerId = res + cb(null, node) + }) + ], callback) +} + +if (!isNode) { + describe('.pubsub-browser (pubsub not supported in the browsers currently)', () => { + const topic = 'pubsub-tests' + + let factory + let ipfs1 + + before((done) => { + factory = new FactoryClient() + + series([ + (cb) => spawnWithId(factory, cb) + ], (err, nodes) => { + if (err) { + return done(err) + } + + ipfs1 = nodes[0] + done() + }) + }) + + after((done) => { + factory.dismantle(done) + }) + + describe('everything errors', () => { + describe('Callback API', () => { + describe('.publish', () => { + it('throws an error if called in the browser', (done) => { + ipfs1.pubsub.publish(topic, 'hello friend', (err, topics) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + }) + }) + }) + + describe('.subscribe', () => { + const handler = () => {} + it('throws an error if called in the browser', (done) => { + ipfs1.pubsub.subscribe(topic, {}, handler, (err, topics) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + }) + }) + }) + + describe('.peers', () => { + it('throws an error if called in the browser', (done) => { + ipfs1.pubsub.peers(topic, (err, topics) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + }) + }) + }) + + describe('.ls', () => { + it('throws an error if called in the browser', (done) => { + ipfs1.pubsub.ls((err, topics) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + }) + }) + }) + }) + + describe('Promise API', () => { + describe('.publish', () => { + it('throws an error if called in the browser', () => { + return ipfs1.pubsub.publish(topic, 'hello friend') + .catch((err) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + }) + }) + }) + + describe('.subscribe', () => { + const handler = () => {} + it('throws an error if called in the browser', (done) => { + ipfs1.pubsub.subscribe(topic, {}, handler) + .catch((err) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + }) + }) + }) + + describe('.peers', () => { + it('throws an error if called in the browser', (done) => { + ipfs1.pubsub.peers(topic) + .catch((err) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + }) + }) + }) + + describe('.ls', () => { + it('throws an error if called in the browser', () => { + return ipfs1.pubsub.ls() + .catch((err) => { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + }) + }) + }) + }) + + describe('.unsubscribe', () => { + it('throws an error if called in the browser', (done) => { + try { + ipfs1.pubsub.unsubscribe() + done('unsubscribe() didn\'t throw an error') + } catch (err) { + expect(err).to.exist() + expect(err.message).to.equal(expectedError) + done() + } + }) + }) + }) + }) +}