chore: adding prettier and improving tooling

Author: JGAntunes

.eslintignore

@@ -0,0 +1 @@
+coverage

.eslintrc

@@ -0,0 +1,14 @@
+
+{
+  "env": {
+    "node": true,
+    "es6": true
+  },
+  "plugins": [
+    "prettier"
+  ],
+  "extends": [
+    "standard",
+    "plugin:prettier/recommended"
+  ]
+}

.prettierrc

@@ -0,0 +1,4 @@
+{
+  "semi": false,
+  "singleQuote": true
+}

package-lock.json

@@ -4,9 +4,37 @@
   "description": "A libp2p compatible pubsub module of the pulsarcast system",
   "main": "src/index.js",
   "scripts": {
+    "fmt": "prettier --write '{,!(coverage)/**/}*.js' && eslint --fix .",
+    "fmt:ci": "prettier -l '{,!(coverage)/**/}*.js'",
+    "lint": "eslint .",
     "test": "nyc --reporter=html --reporter=text mocha --recursive --exit",
-    "test:ci": "npm run lint && npm test",
-    "lint": "standard .  | snazzy"
+    "test:ci": "npm run lint && npm run fmt:ci && npm test"
+  },
+  "nyc": {
+    "exclude": [
+      "test/**",
+      "coverage/*",
+      "npm-debug.log",
+      ".nyc_output"
+    ],
+    "report-dir": "./coverage",
+    "cache": true,
+    "all": true,
+    "extension": [
+      ".js"
+    ]
+  },
+  "husky": {
+    "hooks": {
+      "pre-commit": "lint-staged && npm run test"
+    }
+  },
+  "lint-staged": {
+    "*.js": [
+      "eslint --fix",
+      "prettier --write",
+      "git add"
+    ]
   },
   "pre-commit": [
     "lint",
@@ -49,9 +77,17 @@
     "libp2p-spdy": "^0.13.1",
     "libp2p-tcp": "jgantunes/js-libp2p-tcp",
     "mocha": "^5.2.0",
-    "nyc": "^13.3.0",
-    "pre-commit": "^1.2.2",
-    "snazzy": "^7.1.1",
-    "standard": "^11.0.1"
+    "nyc": "^14.0.0",
+    "eslint": "^5.11.1",
+    "eslint-config-prettier": "^3.3.0",
+    "eslint-config-standard": "^12.0.0",
+    "eslint-plugin-import": "^2.14.0",
+    "eslint-plugin-node": "^8.0.0",
+    "eslint-plugin-prettier": "^3.0.1",
+    "eslint-plugin-promise": "^4.0.1",
+    "eslint-plugin-standard": "^4.0.0",
+    "husky": "^1.3.0",
+    "lint-staged": "^8.1.0",
+    "prettier": "^1.15.3"
   }
 }

package.json

@@ -6,13 +6,10 @@ const dagCBOR = require('ipld-dag-cbor')
 const CID = require('cids')
 
 const config = require('../config')
-const {
-  linkUnmarshalling,
-  linkMarshalling
-} = require('./utils')
+const { linkUnmarshalling, linkMarshalling } = require('./utils')
 
 class EventNode {
-  constructor (topicCID, author, payload, options = {}) {
+  constructor(topicCID, author, payload, options = {}) {
     // TODO check it is a CID maybe?
     assert(topicCID, 'Need a topicCID object to create an event node')
     assert(author, 'Need an author to create an event node')
@@ -27,9 +24,11 @@ class EventNode {
     this.metadata = createMetadata(options.metadata)
   }
 
-  static deserialize (event) {
+  static deserialize(event) {
     const topicCID = linkUnmarshalling(event.topic)
-    const publisher = event.publisher ? PeerId.createFromBytes(event.publisher) : null
+    const publisher = event.publisher
+      ? PeerId.createFromBytes(event.publisher)
+      : null
     const author = PeerId.createFromBytes(event.author)
     const payload = event.payload
     const parent = linkUnmarshalling(event.parent)
@@ -41,18 +40,18 @@ class EventNode {
     })
   }
 
-  static deserializeCBOR (event, cb) {
+  static deserializeCBOR(event, cb) {
     dagCBOR.util.deserialize(event, (err, result) => {
       if (err) return cb(err)
       cb(null, EventNode.deserialize(result))
     })
   }
 
-  get isPublished () {
+  get isPublished() {
     return Boolean(this.publisher)
   }
 
-  getReadableFormat () {
+  getReadableFormat() {
     return {
       topicCID: this.topicCID.toBaseEncodedString(),
       author: this.author.toB58String(),
@@ -64,11 +63,11 @@ class EventNode {
     }
   }
 
-  getCID (cb) {
+  getCID(cb) {
     dagCBOR.util.cid(this.serialize(), cb)
   }
 
-  serialize () {
+  serialize() {
     return {
       topic: linkMarshalling(this.topicCID),
       publisher: this.isPublished ? this.publisher.toBytes() : null,
@@ -82,13 +81,13 @@ class EventNode {
     }
   }
 
-  serializeCBOR (cb) {
+  serializeCBOR(cb) {
     const serialized = this.serialize()
     dagCBOR.util.serialize(serialized, cb)
   }
 }
 
-function createMetadata ({
+function createMetadata({
   created = new Date(),
   protocolVersion = config.protocol
 } = {}) {

src/dag/event-node.js

@@ -7,7 +7,7 @@ const log = require('../utils/logger')
 // TODO right now memory usage grows indefinitely
 class EventTree {
   // TODO probably need access to the DHT?
-  constructor (topicNode) {
+  constructor(topicNode) {
     // TODO check it is a CID maybe?
     assert(topicNode, 'Need a topicNode object to create an event tree')
     log.trace(`New event tree for topic ${topicNode.name}`)
@@ -21,24 +21,22 @@ class EventTree {
     })
   }
 
-  addNew (eventNode, options, cb) {
+  addNew(eventNode, options, cb) {
     const done = cb || options
-    const {parent} = options
+    const { parent } = options
     const eventLinking = this.topicNode.metadata.eventLinking
 
     if (eventLinking === 'custom' && !parent) {
       return done(new Error('Event requires custom parent to be provided'))
     }
 
     // Set the parent link
-    eventNode.parent = eventLinking === 'custom'
-      ? parent
-      : this.mostRecent
+    eventNode.parent = eventLinking === 'custom' ? parent : this.mostRecent
 
     this.add(eventNode, done)
   }
 
-  add (eventNode, cb) {
+  add(eventNode, cb) {
     eventNode.getCID((err, eventCID) => {
       if (err) return cb(err)
 
@@ -59,14 +57,13 @@ class EventTree {
     })
   }
 
-  get (eventCID) {
+  get(eventCID) {
     return this.eventTree.get(eventCID.toBaseEncodedString())
   }
 
   // TODO to should be optional and could be a CID or a number
   // defaults to 1 level of resolution
-  resolve (fromCID, to) {
-  }
+  resolve(fromCID, to) {}
 }
 
 module.exports = EventTree

src/dag/event-tree.js

@@ -6,14 +6,12 @@ const dagCBOR = require('ipld-dag-cbor')
 const CID = require('cids')
 
 const config = require('../config')
-const eventLinkTypes = require('../messages/protobuffers').TopicDescriptor.MetaData.EventLinking
-const {
-  linkUnmarshalling,
-  linkMarshalling
-} = require('./utils')
+const eventLinkTypes = require('../messages/protobuffers').TopicDescriptor
+  .MetaData.EventLinking
+const { linkUnmarshalling, linkMarshalling } = require('./utils')
 
 class TopicNode {
-  constructor (name, author, options = {}) {
+  constructor(name, author, options = {}) {
     assert(author, 'Need an author to create a topic node')
 
     this.name = name
@@ -23,19 +21,21 @@ class TopicNode {
     this.parent = options.parent ? new CID(options.parent) : null
     if (options.subTopics) {
       this.subTopics = Object.entries(options.subTopics)
-        .map(([name, topicB58Str]) => ({[name]: new CID(topicB58Str)}))
-        .reduce((topics, topic) => ({...topic, ...topics}), {})
+        .map(([name, topicB58Str]) => ({ [name]: new CID(topicB58Str) }))
+        .reduce((topics, topic) => ({ ...topic, ...topics }), {})
     }
 
     this.metadata = createMetadata(options.metadata)
   }
 
-  static deserialize (topic) {
+  static deserialize(topic) {
     const author = PeerId.createFromBytes(topic.author)
     const parent = linkUnmarshalling(topic.parent)
-    const subTopics = Object.entries(topic['#']).map(([name, dagLink]) => {
-      return {[name]: linkUnmarshalling(dagLink)}
-    }).reduce((topics, topic) => ({...topic, ...topics}), {})
+    const subTopics = Object.entries(topic['#'])
+      .map(([name, dagLink]) => {
+        return { [name]: linkUnmarshalling(dagLink) }
+      })
+      .reduce((topics, topic) => ({ ...topic, ...topics }), {})
 
     return new TopicNode(topic.name, author, {
       subTopics,
@@ -44,14 +44,14 @@ class TopicNode {
     })
   }
 
-  static deserializeCBOR (topic, cb) {
+  static deserializeCBOR(topic, cb) {
     dagCBOR.util.deserialize(topic, (err, result) => {
       if (err) return cb(err)
       cb(null, TopicNode.deserialize(result))
     })
   }
 
-  getReadableFormat () {
+  getReadableFormat() {
     const allowedPublishers = Array.isArray(this.metadata.allowedPublishers)
       ? this.metadata.allowedPublishers.map(p => p.toB58String())
       : this.metadata.allowedPublishers
@@ -70,40 +70,44 @@ class TopicNode {
     }
   }
 
-  getCID (cb) {
+  getCID(cb) {
     dagCBOR.util.cid(this.serialize(), cb)
   }
 
-  serialize () {
+  serialize() {
     return {
       name: this.name,
       author: this.author.toBytes(),
       parent: linkMarshalling(this.parent),
-      '#': Object.entries(this.subTopics).map(([name, cid]) => {
-        return {[name]: linkMarshalling(cid)}
-      }).reduce((topics, topic) => ({...topic, ...topics}), {}),
+      '#': Object.entries(this.subTopics)
+        .map(([name, cid]) => {
+          return { [name]: linkMarshalling(cid) }
+        })
+        .reduce((topics, topic) => ({ ...topic, ...topics }), {}),
       metadata: serializeMetadata(this.metadata)
     }
   }
 
-  serializeCBOR (cb) {
+  serializeCBOR(cb) {
     const serialized = this.serialize()
     dagCBOR.util.serialize(serialized, cb)
   }
 }
 
-function serializeMetadata (metadata) {
-  const allowedPublishers = {enabled: false, peers: []}
+function serializeMetadata(metadata) {
+  const allowedPublishers = { enabled: false, peers: [] }
   if (metadata.allowedPublishers) {
     allowedPublishers.enabled = true
-    allowedPublishers.peers = metadata.allowedPublishers.map((peer) => peer.toBytes())
+    allowedPublishers.peers = metadata.allowedPublishers.map(peer =>
+      peer.toBytes()
+    )
   }
 
-  const requestToPublish = {enabled: false, peers: []}
+  const requestToPublish = { enabled: false, peers: [] }
   if (metadata.requestToPublish) {
     requestToPublish.enabled = true
     requestToPublish.peers = Array.isArray(metadata.requestToPublish)
-      ? metadata.requestToPublish.map((peer) => peer.toBytes())
+      ? metadata.requestToPublish.map(peer => peer.toBytes())
       : []
   }
 
@@ -116,15 +120,22 @@ function serializeMetadata (metadata) {
   }
 }
 
-function deserializeMetadata (metadata) {
+function deserializeMetadata(metadata) {
   const allowedPublishers = metadata.allowedPublishers.enabled
-    ? metadata.allowedPublishers.peers.map((peer) => PeerId.createFromBytes(peer))
+    ? metadata.allowedPublishers.peers.map(peer => PeerId.createFromBytes(peer))
     : false
 
   let requestToPublish
   if (!metadata.requestToPublish.enabled) requestToPublish = false
-  if (metadata.requestToPublish.enabled && !metadata.requestToPublish.peers.length) requestToPublish = true
-  else requestToPublish = metadata.requestToPublish.peers.map((peer) => PeerId.createFromBytes(peer))
+  if (
+    metadata.requestToPublish.enabled &&
+    !metadata.requestToPublish.peers.length
+  )
+    requestToPublish = true
+  else
+    requestToPublish = metadata.requestToPublish.peers.map(peer =>
+      PeerId.createFromBytes(peer)
+    )
 
   return {
     ...metadata,
@@ -136,7 +147,7 @@ function deserializeMetadata (metadata) {
   }
 }
 
-function createMetadata ({
+function createMetadata({
   allowedPublishers = false,
   requestToPublish = true,
   eventLinking = 'LAST_SEEN',

src/dag/topic-node.js

@@ -9,11 +9,11 @@
 // TODO right now memory usage grows indefinitely
 class TopicTree {
   // TODO probably need access to the DHT?
-  constructor () {
+  constructor() {
     this.topicTree = new Map()
   }
 
-  add (topicNode, cb) {
+  add(topicNode, cb) {
     topicNode.getCID((err, topicCID) => {
       if (err) return cb(err)
 
@@ -22,14 +22,13 @@ class TopicTree {
     })
   }
 
-  get (topicCID) {
+  get(topicCID) {
     return this.eventTree.get(topicCID.toBaseEncodedString())
   }
 
   // TODO to should be optional and could be a CID or a number
   // defaults to 1 level of resolution
-  resolve (fromCID, to) {
-  }
+  resolve(fromCID, to) {}
 }
 
 module.exports = TopicTree

src/dag/topic-tree.js

@@ -2,19 +2,19 @@
 
 const CID = require('cids')
 
-function linkMarshalling (link) {
+function linkMarshalling(link) {
   // No link
   if (!link) return {}
   // It's a CID already
-  if (CID.isCID(link)) return {'/': link.buffer}
+  if (CID.isCID(link)) return { '/': link.buffer }
   // It's an object but empty
   if (!link['/'] && typeof link === 'object') return {}
   // It can be a link already or just the multihash
   const newCID = new CID(link['/'] || link)
-  return {'/': newCID.buffer}
+  return { '/': newCID.buffer }
 }
 
-function linkUnmarshalling (link) {
+function linkUnmarshalling(link) {
   // Link is already a CID
   // happens when dagcbor utils is used directly
   if (CID.isCID(link)) return link

src/dag/utils.js

@@ -6,7 +6,7 @@ const config = require('../config')
 
 // Update RPC message will handle neighbourhood
 // updates
-function update (topic, {parents, children}) {
+function update(topic, { parents, children }) {
   const metadata = createMetadata()
   return {
     op: 'UPDATE',
@@ -19,47 +19,47 @@ function update (topic, {parents, children}) {
   }
 }
 
-function publish (eventNode) {
+function publish(eventNode) {
   return {
     op: 'PUBLISH_EVENT',
     metadata: createMetadata(),
     event: eventNode
   }
 }
 
-function requestToPublish (eventNode) {
+function requestToPublish(eventNode) {
   return {
     op: 'REQUEST_TO_PUBLISH',
     metadata: createMetadata(),
     event: eventNode
   }
 }
 
-function joinTopic (topic) {
+function joinTopic(topic) {
   return {
     op: 'JOIN_TOPIC',
     topicId: new CID(topic),
     metadata: createMetadata()
   }
 }
 
-function leaveTopic (topic) {
+function leaveTopic(topic) {
   return {
     op: 'LEAVE_TOPIC',
     topicId: new CID(topic),
     metadata: createMetadata()
   }
 }
 
-function newTopic (topicNode) {
+function newTopic(topicNode) {
   return {
     op: 'NEW_TOPIC',
     topic: topicNode,
     metadata: createMetadata()
   }
 }
 
-function createMetadata () {
+function createMetadata() {
   const now = new Date()
   return {
     protocolVersion: config.protocol,

src/index.js

@@ -7,7 +7,7 @@ const EventNode = require('../dag/event-node')
 const TopicNode = require('../dag/topic-node')
 const ops = require('./protobuffers').RPC.Operation
 
-function unmarshall (message) {
+function unmarshall(message) {
   // TODO improve code unmarshalling
   const result = {}
   result.metadata = message.metadata
@@ -41,7 +41,7 @@ function unmarshall (message) {
   return result
 }
 
-function marshall (message) {
+function marshall(message) {
   // TODO improve code marshalling
   const result = {}
   result.metadata = message.metadata

src/messages/create-rpc.js

@@ -2,22 +2,32 @@
 
 const Joi = require('joi')
 
-const metadata = Joi.object().keys({
-  created: Joi.date().iso().required(),
-  protocolVersion: Joi.string().required()
-  // signature: Joi.binary()
-}).required()
+const metadata = Joi.object()
+  .keys({
+    created: Joi.date()
+      .iso()
+      .required(),
+    protocolVersion: Joi.string().required()
+    // signature: Joi.binary()
+  })
+  .required()
 
 const eventDescriptor = Joi.object().keys({
-  publisher: Joi.binary().required().allow(null),
+  publisher: Joi.binary()
+    .required()
+    .allow(null),
   author: Joi.binary().required(),
-  parent: Joi.object().keys({
-    '/': Joi.binary().allow(null)
-  }).required(),
+  parent: Joi.object()
+    .keys({
+      '/': Joi.binary().allow(null)
+    })
+    .required(),
   payload: Joi.binary().required(),
-  topic: Joi.object().keys({
-    '/': Joi.binary().required()
-  }).required(),
+  topic: Joi.object()
+    .keys({
+      '/': Joi.binary().required()
+    })
+    .required(),
   metadata
 })
 

src/messages/marshalling.js

@@ -4,8 +4,12 @@ const Joi = require('joi')
 
 const peerTree = Joi.object().keys({
   topic: Joi.binary().required(),
-  parents: Joi.array().items(Joi.binary()).required(),
-  children: Joi.array().items(Joi.binary()).required()
+  parents: Joi.array()
+    .items(Joi.binary())
+    .required(),
+  children: Joi.array()
+    .items(Joi.binary())
+    .required()
 })
 
 module.exports = peerTree

src/messages/schemas/event-descriptor.js

@@ -6,18 +6,25 @@ const eventDescriptor = require('./event-descriptor')
 const peerTree = require('./peer-tree')
 const ops = require('../protobuffers').RPC.Operation
 
-const metadata = Joi.object().keys({
-  created: Joi.date().iso(),
-  protocolVersion: Joi.string()
-}).required()
+const metadata = Joi.object()
+  .keys({
+    created: Joi.date().iso(),
+    protocolVersion: Joi.string()
+  })
+  .required()
 
-const rpc = Joi.object().keys({
-  op: Joi.number().integer().valid(Object.values(ops)).required(),
-  metadata,
-  topicId: Joi.binary(),
-  event: eventDescriptor,
-  topic: topicDescriptor,
-  peerTree: peerTree
-}).required()
+const rpc = Joi.object()
+  .keys({
+    op: Joi.number()
+      .integer()
+      .valid(Object.values(ops))
+      .required(),
+    metadata,
+    topicId: Joi.binary(),
+    event: eventDescriptor,
+    topic: topicDescriptor,
+    peerTree: peerTree
+  })
+  .required()
 
 module.exports = rpc

src/messages/schemas/peer-tree.js

@@ -1,33 +1,55 @@
 'use strict'
 
 const Joi = require('joi')
-const eventLinking = require('../protobuffers').TopicDescriptor.MetaData.EventLinking
+const eventLinking = require('../protobuffers').TopicDescriptor.MetaData
+  .EventLinking
 
-const metadata = Joi.object().keys({
-  created: Joi.date().iso().required(),
-  protocolVersion: Joi.string().required(),
-  allowedPublishers: Joi.object().keys({
-    enabled: Joi.boolean(),
-    peers: Joi.alternatives()
-      .when('enabled', {is: true, then: Joi.array().items(Joi.binary()).min(1)})
-  }).required(),
-  requestToPublish: Joi.object().keys({
-    enabled: Joi.boolean(),
-    peers: Joi.array().items(Joi.binary())
-  }).required(),
-  eventLinking: Joi.string().valid(Object.values(eventLinking)).required()
-  // signature: Joi.binary()
-}).required()
+const metadata = Joi.object()
+  .keys({
+    created: Joi.date()
+      .iso()
+      .required(),
+    protocolVersion: Joi.string().required(),
+    allowedPublishers: Joi.object()
+      .keys({
+        enabled: Joi.boolean(),
+        peers: Joi.alternatives().when('enabled', {
+          is: true,
+          then: Joi.array()
+            .items(Joi.binary())
+            .min(1)
+        })
+      })
+      .required(),
+    requestToPublish: Joi.object()
+      .keys({
+        enabled: Joi.boolean(),
+        peers: Joi.array().items(Joi.binary())
+      })
+      .required(),
+    eventLinking: Joi.string()
+      .valid(Object.values(eventLinking))
+      .required()
+    // signature: Joi.binary()
+  })
+  .required()
 
 const topicDescriptor = Joi.object().keys({
   name: Joi.string().required(),
   author: Joi.binary().required(),
-  parent: Joi.object().keys({
-    '/': Joi.binary()
-  }).required(),
-  '#': Joi.object().pattern(Joi.string(), Joi.object().keys({
-    '/': Joi.binary().required()
-  })).required(),
+  parent: Joi.object()
+    .keys({
+      '/': Joi.binary()
+    })
+    .required(),
+  '#': Joi.object()
+    .pattern(
+      Joi.string(),
+      Joi.object().keys({
+        '/': Joi.binary().required()
+      })
+    )
+    .required(),
   metadata
 })
 

src/messages/schemas/rpc.js

@@ -9,8 +9,8 @@ const assert = require('assert')
 const log = require('./utils/logger')
 
 class Peer extends EventEmitter {
-  constructor (peerInfo, conn = null) {
-    log.trace('New peer registered %j', {peer: peerInfo.id.toB58String()})
+  constructor(peerInfo, conn = null) {
+    log.trace('New peer registered %j', { peer: peerInfo.id.toB58String() })
     assert(peerInfo, 'Need a peerInfo object to initiate the peer')
     super()
 
@@ -24,11 +24,11 @@ class Peer extends EventEmitter {
     }
   }
 
-  isConnected () {
+  isConnected() {
     return !!this.conn
   }
 
-  attachConnection (conn) {
+  attachConnection(conn) {
     if (this.conn) {
       // TODO close previously existing connection
     }
@@ -50,31 +50,33 @@ class Peer extends EventEmitter {
     this.emit('connection')
   }
 
-  sendMessages (messages) {
-    log.trace('Pushing message to peer %j', {peer: this.info.id.toB58String()})
+  sendMessages(messages) {
+    log.trace('Pushing message to peer %j', {
+      peer: this.info.id.toB58String()
+    })
     this.stream.push(messages)
   }
 
-  updateTree (topic, {parents = [], children = []}) {
-    this.trees.set(topic, {parents, children})
+  updateTree(topic, { parents = [], children = [] }) {
+    this.trees.set(topic, { parents, children })
   }
 
-  removeTree (topic) {
+  removeTree(topic) {
     const tree = this.trees.get(topic)
     if (!tree) return
 
     this.trees.delete(topic)
     return tree
   }
 
-  addChildren (topic, children) {
+  addChildren(topic, children) {
     const tree = this.trees.get(topic)
     if (!tree) {
-      this.trees.set(topic, {children, parents: []})
+      this.trees.set(topic, { children, parents: [] })
       return
     }
-    children.forEach((child) => {
-      const exists = tree.children.find((peer) => {
+    children.forEach(child => {
+      const exists = tree.children.find(peer => {
         return peer.info.id.isEqual(child.info.id)
       })
       if (!exists) {
@@ -83,14 +85,14 @@ class Peer extends EventEmitter {
     })
   }
 
-  addParents (topic, parents) {
+  addParents(topic, parents) {
     const tree = this.trees.get(topic)
     if (!tree) {
-      this.trees.set(topic, {parents, children: []})
+      this.trees.set(topic, { parents, children: [] })
       return
     }
-    parents.forEach((parent) => {
-      const exists = tree.parents.find((peer) => {
+    parents.forEach(parent => {
+      const exists = tree.parents.find(peer => {
         return peer.info.id.isEqual(parent.info.id)
       })
       if (!exists) {
@@ -99,18 +101,18 @@ class Peer extends EventEmitter {
     })
   }
 
-  removePeer (topic, peerId) {
+  removePeer(topic, peerId) {
     const tree = this.trees.get(topic)
     const newParents = []
     const newChildren = []
     if (!tree) return
 
-    tree.parents.forEach((parent) => {
+    tree.parents.forEach(parent => {
       if (parent.info.id.isEqual(peerId)) return
       newParents.push(parent)
     })
 
-    tree.children.forEach((child) => {
+    tree.children.forEach(child => {
       if (child.info.id.isEqual(peerId)) return
       newChildren.push(child)
     })
@@ -119,7 +121,7 @@ class Peer extends EventEmitter {
     tree.parents = newParents
   }
 
-  close (callback) {
+  close(callback) {
     // End the pushable
     if (this.stream) {
       this.stream.end()
@@ -135,7 +137,7 @@ class Peer extends EventEmitter {
 
   // Only close the connection if no other
   // topic subscriptions are kept by this peer
-  gracefulClose (callback) {
+  gracefulClose(callback) {
     if (this.trees.size === 0) this.close(callback)
   }
 }

src/messages/schemas/topic-descriptor.js

@@ -5,7 +5,7 @@ const receive = require('./receive')
 
 module.exports = rpc
 
-function rpc (pulsarcastNode) {
+function rpc(pulsarcastNode) {
   return {
     send: send(pulsarcastNode),
     receive: receive(pulsarcastNode)

src/peer.js

@@ -9,7 +9,7 @@ const { protobuffers, schemas, marshalling } = require('../messages')
 
 const ops = protobuffers.RPC.Operation
 
-function createRPCHandlers (pulsarcastNode) {
+function createRPCHandlers(pulsarcastNode) {
   const dht = pulsarcastNode.libp2p._dht
 
   return {
@@ -25,10 +25,10 @@ function createRPCHandlers (pulsarcastNode) {
     genericHandler
   }
 
-  function publish (idB58Str, eventNode, callback) {
+  function publish(idB58Str, eventNode, callback) {
     // Only consider the message if we have data
     if (!eventNode) return callback()
-    const {me} = pulsarcastNode
+    const { me } = pulsarcastNode
     const myId = me.info.id
 
     // Publish is from this node so it's a new event
@@ -38,67 +38,94 @@ function createRPCHandlers (pulsarcastNode) {
     getTopic(dht, eventNode.topicCID, (err, topicNode) => {
       if (err) return callback(err)
 
-      const {
-        allowedPublishers
-      } = topicNode.metadata
+      const { allowedPublishers } = topicNode.metadata
 
       // New event published at this node and not allowed to publish
-      if (newEvent && allowedPublishers && !allowedPublishers.find((peer) => peer.isEqual(myId))) {
-        return requestToPublish(myId.toB58String(), eventNode, (err, topicNode, eventNode) => {
-          callback(err, null, topicNode, eventNode)
-        })
+      if (
+        newEvent &&
+        allowedPublishers &&
+        !allowedPublishers.find(peer => peer.isEqual(myId))
+      ) {
+        return requestToPublish(
+          myId.toB58String(),
+          eventNode,
+          (err, topicNode, eventNode) => {
+            callback(err, null, topicNode, eventNode)
+          }
+        )
       }
 
       // TODO check publisher is allowed
 
       const options = { isNewEvent: newEvent }
-      pulsarcastNode.rpc.send.event.publish(topicNode, eventNode, idB58Str, options, callback)
+      pulsarcastNode.rpc.send.event.publish(
+        topicNode,
+        eventNode,
+        idB58Str,
+        options,
+        callback
+      )
     })
   }
 
-  function requestToPublish (idB58Str, eventNode, callback) {
+  function requestToPublish(idB58Str, eventNode, callback) {
     // Only consider the message if we have data
     if (!eventNode) return
-    const {me} = pulsarcastNode
+    const { me } = pulsarcastNode
     const myId = me.info.id
 
     log.trace('Got request to publish  %j', eventNode)
 
     getTopic(dht, eventNode.topicCID, (err, topicNode) => {
       if (err) return callback(err)
 
-      const {
-        allowedPublishers,
-        requestToPublish
-      } = topicNode.metadata
+      const { allowedPublishers, requestToPublish } = topicNode.metadata
 
       // TODO better handling for this case
       if (!requestToPublish) return callback(null, topicNode, eventNode)
       // TODO better handling for this case
-      if (Array.isArray(requestToPublish) && !requestToPublish.find((peer) => peer.isEqual(topicNode.author))) return callback(null, topicNode, eventNode)
+      if (
+        Array.isArray(requestToPublish) &&
+        !requestToPublish.find(peer => peer.isEqual(topicNode.author))
+      )
+        return callback(null, topicNode, eventNode)
 
       // Publish if I'm allowed to
-      if (!allowedPublishers || allowedPublishers.find((peer) => peer.isEqual(myId))) {
-        return pulsarcastNode.rpc.send.event.publish(topicNode, eventNode, myId.toB58String(), {isNewEvent: true}, callback)
+      if (
+        !allowedPublishers ||
+        allowedPublishers.find(peer => peer.isEqual(myId))
+      ) {
+        return pulsarcastNode.rpc.send.event.publish(
+          topicNode,
+          eventNode,
+          myId.toB58String(),
+          { isNewEvent: true },
+          callback
+        )
       }
       // Propagate request to publish
-      pulsarcastNode.rpc.send.event.requestToPublish(topicNode, eventNode, idB58Str, callback)
+      pulsarcastNode.rpc.send.event.requestToPublish(
+        topicNode,
+        eventNode,
+        idB58Str,
+        callback
+      )
     })
   }
 
-  function update (idB58Str, peerTree) {
+  function update(idB58Str, peerTree) {
     // Only consider the emessage if we have data
     if (!peerTree) return
     const topicCID = peerTree.topicId.toBaseEncodedString()
 
     log.trace('Got update  %j', peerTree)
 
-    const {peers} = pulsarcastNode
+    const { peers } = pulsarcastNode
     peers.get(idB58Str).updateTree(topicCID, peerTree)
   }
 
-  function join (idB58Str, topicCID, callback) {
-    const {me, peers} = pulsarcastNode
+  function join(idB58Str, topicCID, callback) {
+    const { me, peers } = pulsarcastNode
     const topicB58Str = topicCID.toBaseEncodedString()
 
     // The peer should already be in the list given that
@@ -117,17 +144,19 @@ function createRPCHandlers (pulsarcastNode) {
         child.addParents(topicB58Str, [me])
         // TODO take care of delivering initial state
         // This node is the root node for the topic
-        if (me.info.id.isEqual(topicNode.author)) return callback(null, topicNode)
+        if (me.info.id.isEqual(topicNode.author))
+          return callback(null, topicNode)
         // Check if we have a set of parents for this topic
-        if (me.trees.get(topicB58Str).parents > 0) return callback(null, topicNode)
+        if (me.trees.get(topicB58Str).parents > 0)
+          return callback(null, topicNode)
       }
 
       pulsarcastNode.rpc.send.topic.join(topicNode, callback)
     })
   }
 
-  function leave (idB58Str, topicCID, callback) {
-    const {me, peers} = pulsarcastNode
+  function leave(idB58Str, topicCID, callback) {
+    const { me, peers } = pulsarcastNode
     const topicB58Str = topicCID.toBaseEncodedString()
     const myId = me.info.id
 
@@ -154,9 +183,14 @@ function createRPCHandlers (pulsarcastNode) {
 
         // Need to forward the leave rpc
         const peers = tree.children.concat(tree.parents)
-        return eachLimit(peers, 20, (peer, done) => {
-          pulsarcastNode.rpc.send.topic.leave(topicNode, peer, done)
-        }, err => callback(err))
+        return eachLimit(
+          peers,
+          20,
+          (peer, done) => {
+            pulsarcastNode.rpc.send.topic.leave(topicNode, peer, done)
+          },
+          err => callback(err)
+        )
       }
 
       // Remove the peer from our tree
@@ -170,14 +204,18 @@ function createRPCHandlers (pulsarcastNode) {
         if (err) return callback(err)
 
         // If this node is a child, we need to rejoin the topic
-        if (me.trees.get(topicB58Str).children.find((peer) => peer.info.id.isEqual(myId))) {
+        if (
+          me.trees
+            .get(topicB58Str)
+            .children.find(peer => peer.info.id.isEqual(myId))
+        ) {
           pulsarcastNode.rpc.send.topic.join(topicNode, callback)
         }
       })
     })
   }
 
-  function genericHandler (idB58Str, message) {
+  function genericHandler(idB58Str, message) {
     const result = Joi.validate(message, schemas.rpc, {
       abortEarly: true,
       allowUnknown: false,
@@ -191,8 +229,12 @@ function createRPCHandlers (pulsarcastNode) {
     // with type coercion
     const jsonMessage = marshalling.unmarshall(result.value)
 
-    log.trace('Received rpc %j', {handler: 'in', op: jsonMessage.op, from: idB58Str})
-    const errorHandler = (err) => {
+    log.trace('Received rpc %j', {
+      handler: 'in',
+      op: jsonMessage.op,
+      from: idB58Str
+    })
+    const errorHandler = err => {
       if (err) log.error('%j', err)
     }
 

src/rpc/index.js

@@ -9,7 +9,7 @@ const { closestPeerToPeer, store } = require('../utils/dht-helpers')
 
 const RPC = protobuffers.RPC
 
-function createRPCHandlers (pulsarcastNode) {
+function createRPCHandlers(pulsarcastNode) {
   const dht = pulsarcastNode.libp2p._dht
 
   return {
@@ -24,13 +24,13 @@ function createRPCHandlers (pulsarcastNode) {
     }
   }
 
-  function publish (topicNode, eventNode, fromIdB58Str, options, callback) {
+  function publish(topicNode, eventNode, fromIdB58Str, options, callback) {
     if (!callback) {
       callback = options
       options = {}
     }
 
-    const {me} = pulsarcastNode
+    const { me } = pulsarcastNode
     const myId = me.info.id
     const isNewEvent = options.isNewEvent
 
@@ -39,43 +39,52 @@ function createRPCHandlers (pulsarcastNode) {
       eventNode.publisher = myId
     }
 
-    waterfall([
-      topicNode.getCID.bind(topicNode),
-      (topicCID, cb) => {
-        addEvent(topicCID, topicNode, eventNode, {createLink: isNewEvent}, cb)
-      },
-      (linkedEvent, cb) => {
-        if (!isNewEvent) return cb(null, null, linkedEvent)
-
-        // Publish is being created at this node, not just forwardind,
-        // so add it to DHT and propagate it through our whole topic tree
-        store(dht, linkedEvent, cb)
-      }
-    ], (err, eventCID, linkedEvent) => {
-      if (err) return callback(err)
-
-      const topicB58Str = linkedEvent.topicCID.toBaseEncodedString()
-      const rpc = createRPC.event.publish(linkedEvent)
-      const trees = pulsarcastNode.me.trees.get(topicB58Str)
+    waterfall(
+      [
+        topicNode.getCID.bind(topicNode),
+        (topicCID, cb) => {
+          addEvent(
+            topicCID,
+            topicNode,
+            eventNode,
+            { createLink: isNewEvent },
+            cb
+          )
+        },
+        (linkedEvent, cb) => {
+          if (!isNewEvent) return cb(null, null, linkedEvent)
+
+          // Publish is being created at this node, not just forwardind,
+          // so add it to DHT and propagate it through our whole topic tree
+          store(dht, linkedEvent, cb)
+        }
+      ],
+      (err, eventCID, linkedEvent) => {
+        if (err) return callback(err)
 
-      // We're subscribed to this topic, emit the message
-      if (pulsarcastNode.subscriptions.has(topicB58Str)) {
-        pulsarcastNode.emit(topicB58Str, linkedEvent)
+        const topicB58Str = linkedEvent.topicCID.toBaseEncodedString()
+        const rpc = createRPC.event.publish(linkedEvent)
+        const trees = pulsarcastNode.me.trees.get(topicB58Str)
+
+        // We're subscribed to this topic, emit the message
+        if (pulsarcastNode.subscriptions.has(topicB58Str)) {
+          pulsarcastNode.emit(topicB58Str, linkedEvent)
+        }
+        // TODO handle publishing to an event we're not subscribed to
+        if (!trees) return callback(null, eventCID, topicNode, linkedEvent)
+        const { parents, children } = trees
+
+        const peers = [...parents, ...children]
+        peers.forEach(peer => {
+          // Don't forward the message back
+          if (peer.info.id.toB58String() !== fromIdB58Str) send(peer, rpc)
+          return callback(null, eventCID, topicNode, linkedEvent)
+        })
       }
-      // TODO handle publishing to an event we're not subscribed to
-      if (!trees) return callback(null, eventCID, topicNode, linkedEvent)
-      const { parents, children } = trees
-
-      const peers = [...parents, ...children]
-      peers.forEach(peer => {
-        // Don't forward the message back
-        if (peer.info.id.toB58String() !== fromIdB58Str) send(peer, rpc)
-        return callback(null, eventCID, topicNode, linkedEvent)
-      })
-    })
+    )
   }
 
-  function requestToPublish (topicNode, eventNode, fromIdB58Str, callback) {
+  function requestToPublish(topicNode, eventNode, fromIdB58Str, callback) {
     const rpc = createRPC.event.requestToPublish(eventNode)
 
     topicNode.getCID((err, topicCID) => {
@@ -98,30 +107,33 @@ function createRPCHandlers (pulsarcastNode) {
 
   // Join finds the closest peer to the topic CID
   // and sends the rpc join message
-  function joinTopic (topicNode, callback) {
+  function joinTopic(topicNode, callback) {
     const { me } = pulsarcastNode
     topicNode.getCID((err, topicCID) => {
       if (err) return callback(err)
 
       const rpc = createRPC.topic.join(topicCID)
       // Get the closest peer to the topic author stored locally
-      waterfall([
-        closestPeerToPeer.bind(null, dht, topicNode.author),
-        pulsarcastNode._getPeer.bind(pulsarcastNode)
-      ], (err, peer) => {
-        if (err) return callback(err)
-        // Add peer to my tree
-        me.addParents(topicCID.toBaseEncodedString(), [peer])
-        // Add me to peer's tree
-        peer.addChildren(topicCID.toBaseEncodedString(), [me])
-        send(peer, rpc)
-
-        callback(null, topicNode)
-      })
+      waterfall(
+        [
+          closestPeerToPeer.bind(null, dht, topicNode.author),
+          pulsarcastNode._getPeer.bind(pulsarcastNode)
+        ],
+        (err, peer) => {
+          if (err) return callback(err)
+          // Add peer to my tree
+          me.addParents(topicCID.toBaseEncodedString(), [peer])
+          // Add me to peer's tree
+          peer.addChildren(topicCID.toBaseEncodedString(), [me])
+          send(peer, rpc)
+
+          callback(null, topicNode)
+        }
+      )
     })
   }
 
-  function leaveTopic (topicNode, toPeer, callback) {
+  function leaveTopic(topicNode, toPeer, callback) {
     topicNode.getCID((err, topicCID) => {
       if (err) return callback(err)
 
@@ -132,7 +144,7 @@ function createRPCHandlers (pulsarcastNode) {
   }
 
   // TODO for now only store topic descriptor
-  function newTopic (topicNode, options, callback) {
+  function newTopic(topicNode, options, callback) {
     // check if options exist
     if (!callback) {
       callback = options
@@ -142,19 +154,23 @@ function createRPCHandlers (pulsarcastNode) {
     store(dht, topicNode, callback)
   }
 
-  function send (peer, rpc) {
-    log.trace('Sending rpc %j', {handler: 'out', op: rpc.op, to: peer.info.id.toB58String()})
+  function send(peer, rpc) {
+    log.trace('Sending rpc %j', {
+      handler: 'out',
+      op: rpc.op,
+      to: peer.info.id.toB58String()
+    })
 
     const rpcToSend = marshalling.marshall(rpc)
-    const encodedMessage = RPC.encode({msgs: [rpcToSend]})
+    const encodedMessage = RPC.encode({ msgs: [rpcToSend] })
 
     peer.sendMessages(encodedMessage)
   }
 
   // Helper funcs
-  function addEvent (topicCID, topicNode, eventNode, {createLink}, cb) {
+  function addEvent(topicCID, topicNode, eventNode, { createLink }, cb) {
     const topicB58Str = topicCID.toBaseEncodedString()
-    const {eventTrees} = pulsarcastNode
+    const { eventTrees } = pulsarcastNode
     let eventTree
 
     // Add event tree if it does not exist

src/rpc/receive.js

@@ -8,42 +8,46 @@ const log = require('./logger')
 const TopicNode = require('../dag/topic-node')
 const EventNode = require('../dag/event-node')
 
-function closestPeerToPeer (dht, peerId, cb) {
+function closestPeerToPeer(dht, peerId, cb) {
   convertPeerId(peerId, (err, key) => {
     if (err) cb(err)
     cb(null, dht.routingTable.closestPeer(key))
   })
 }
 
 // Store a event node or topic node
-function store (dht, dagNode, cb) {
-  waterfall([
-    (done) => parallel([
-      dagNode.getCID.bind(dagNode),
-      dagNode.serializeCBOR.bind(dagNode)
-    ], done),
-    ([cid, serialized], done) => {
-      log.trace('Storing node %j', {
-        cid: cid.toBaseEncodedString(),
-        ...dagNode.getReadableFormat()
-      })
-      dht.put(cid.buffer, serialized, (err) => done(err, cid))
-    }
-  ], (err, cid) => cb(err, cid, dagNode))
+function store(dht, dagNode, cb) {
+  waterfall(
+    [
+      done =>
+        parallel(
+          [dagNode.getCID.bind(dagNode), dagNode.serializeCBOR.bind(dagNode)],
+          done
+        ),
+      ([cid, serialized], done) => {
+        log.trace('Storing node %j', {
+          cid: cid.toBaseEncodedString(),
+          ...dagNode.getReadableFormat()
+        })
+        dht.put(cid.buffer, serialized, err => done(err, cid))
+      }
+    ],
+    (err, cid) => cb(err, cid, dagNode)
+  )
 }
 
-function getTopic (dht, topicCID, cb) {
-  waterfall([
-    dht.get.bind(dht, topicCID.buffer, null),
-    TopicNode.deserializeCBOR
-  ], cb)
+function getTopic(dht, topicCID, cb) {
+  waterfall(
+    [dht.get.bind(dht, topicCID.buffer, null), TopicNode.deserializeCBOR],
+    cb
+  )
 }
 
-function getEvent (dht, eventCID, cb) {
-  waterfall([
-    dht.get.bind(dht, eventCID.buffer, null),
-    EventNode.deserializeCBOR
-  ], cb)
+function getEvent(dht, eventCID, cb) {
+  waterfall(
+    [dht.get.bind(dht, eventCID.buffer, null), EventNode.deserializeCBOR],
+    cb
+  )
 }
 
 module.exports = {

src/rpc/send.js

@@ -17,27 +17,27 @@ describe('2 nodes', () => {
   let topic
   let topicCID
 
-  before((done) => {
+  before(done => {
     createNodes(2, (err, p2pNodes) => {
       expect(err).not.to.exist
-      nodes = p2pNodes.map((node) => new Pulsarcast(node))
+      nodes = p2pNodes.map(node => new Pulsarcast(node))
       done()
     })
   })
 
-  it('starts both nodes', (done) => {
-    parallel([
-      nodes[0].start.bind(nodes[0]),
-      nodes[1].start.bind(nodes[1])
-    ], (err) => {
-      expect(err).to.not.exist
-      expect(nodes[0].started).to.be.true
-      expect(nodes[1].started).to.be.true
-      done()
-    })
+  it('starts both nodes', done => {
+    parallel(
+      [nodes[0].start.bind(nodes[0]), nodes[1].start.bind(nodes[1])],
+      err => {
+        expect(err).to.not.exist
+        expect(nodes[0].started).to.be.true
+        expect(nodes[1].started).to.be.true
+        done()
+      }
+    )
   })
 
-  it('creates a topic', (done) => {
+  it('creates a topic', done => {
     nodes[0].createTopic('test', (err, savedCID, topicNode) => {
       expect(err).to.not.exist
       expect(topicNode).to.be.an.instanceof(TopicNode)
@@ -56,26 +56,31 @@ describe('2 nodes', () => {
     })
   })
 
-  it('creates a new topic with a parent', (done) => {
-    nodes[0].createTopic('test-2.0', {parent: topicCID.toBaseEncodedString()}, (err, savedCID, childTopicNode) => {
-      expect(err).to.not.exist
-      expect(childTopicNode).to.be.an.instanceof(TopicNode)
-      expect(childTopicNode.subTopics.meta).to.be.an.instanceof(CID)
-      expect(childTopicNode.parent.equals(topicCID)).to.be.true
-      childTopicNode.getCID((err, cid) => {
+  it('creates a new topic with a parent', done => {
+    nodes[0].createTopic(
+      'test-2.0',
+      { parent: topicCID.toBaseEncodedString() },
+      (err, savedCID, childTopicNode) => {
         expect(err).to.not.exist
-        const topicB58Str = cid.toBaseEncodedString()
-        expect(cid.equals(savedCID)).to.be.true
-        expect(nodes[0].subscriptions.size).to.equal(4)
-        expect(nodes[0].subscriptions.has(topicB58Str)).to.be.true
-        done()
-      })
-    })
+        expect(childTopicNode).to.be.an.instanceof(TopicNode)
+        expect(childTopicNode.subTopics.meta).to.be.an.instanceof(CID)
+        expect(childTopicNode.parent.equals(topicCID)).to.be.true
+        childTopicNode.getCID((err, cid) => {
+          expect(err).to.not.exist
+          const topicB58Str = cid.toBaseEncodedString()
+          expect(cid.equals(savedCID)).to.be.true
+          expect(nodes[0].subscriptions.size).to.equal(4)
+          expect(nodes[0].subscriptions.has(topicB58Str)).to.be.true
+          done()
+        })
+      }
+    )
   })
 
-  it('creates a new topic with a subTopic', (done) => {
-    nodes[0].createTopic('test-with-subtopic',
-      {subTopics: {'some-topic': topicCID.toBaseEncodedString()}},
+  it('creates a new topic with a subTopic', done => {
+    nodes[0].createTopic(
+      'test-with-subtopic',
+      { subTopics: { 'some-topic': topicCID.toBaseEncodedString() } },
       (err, savedCID, newTopicNode) => {
         expect(err).to.not.exist
         expect(newTopicNode.subTopics['some-topic'].equals(topicCID)).to.be.true
@@ -89,28 +94,30 @@ describe('2 nodes', () => {
           expect(nodes[0].subscriptions.has(topicB58Str)).to.be.true
           done()
         })
-      })
+      }
+    )
   })
 
-  it('subscribes to the previously created topic', (done) => {
+  it('subscribes to the previously created topic', done => {
     const topicB58Str = topicCID.toBaseEncodedString()
 
     nodes[1].subscribe(topicB58Str, (err, topicNode) => {
       expect(err).to.not.exist
       expect(topicNode).to.be.an.instanceof(TopicNode)
       expect(nodes[1].subscriptions.size).to.equal(1)
-      expect(nodes[1].subscriptions.has(topicCID.toBaseEncodedString())).to.be.true
+      expect(nodes[1].subscriptions.has(topicCID.toBaseEncodedString())).to.be
+        .true
       expect(topicNode.serialize()).to.deep.equal(topic.serialize())
 
       // Check tree
-      const {parents} = nodes[1].me.trees.get(topicB58Str)
+      const { parents } = nodes[1].me.trees.get(topicB58Str)
       expect(parents).to.have.lengthOf.above(0)
       expect(parents[0].trees.get(topicB58Str).children).to.include(nodes[1].me)
       done()
     })
   })
 
-  it('publishes a message from the non author node', (done) => {
+  it('publishes a message from the non author node', done => {
     const topicB58Str = topicCID.toBaseEncodedString()
     const message = 'foobar'
     // Helper func to run all the expects
@@ -121,7 +128,7 @@ describe('2 nodes', () => {
     }
     let firstEventNode
     // Event listener
-    const listener = (eventNode) => {
+    const listener = eventNode => {
       // Compare serializes of both events
       if (!firstEventNode) {
         firstEventNode = eventNode
@@ -144,30 +151,36 @@ describe('2 nodes', () => {
     nodes[1].once(topicB58Str, listener)
     nodes[0].once(topicB58Str, listener)
 
-    nodes[1].publish(topicB58Str, message, (err, eventCID, topicNode, eventNode) => {
-      expect(err).to.not.exist
-      expect(eventNode).to.be.an.instanceof(EventNode)
-      expect(topicNode).to.be.an.instanceof(TopicNode)
-      expect(eventNode.topicCID.equals(topicCID)).to.be.true
-      expect(eventNode.payload.toString()).to.be.equal(message)
-      expect(eventNode.author.isEqual(nodes[1].me.info.id)).to.be.true
-      // Should be a request to publish
-      expect(eventCID).to.be.null
-      expect(eventNode.isPublished).to.be.false
-      checkAllDone()
-    })
+    nodes[1].publish(
+      topicB58Str,
+      message,
+      (err, eventCID, topicNode, eventNode) => {
+        expect(err).to.not.exist
+        expect(eventNode).to.be.an.instanceof(EventNode)
+        expect(topicNode).to.be.an.instanceof(TopicNode)
+        expect(eventNode.topicCID.equals(topicCID)).to.be.true
+        expect(eventNode.payload.toString()).to.be.equal(message)
+        expect(eventNode.author.isEqual(nodes[1].me.info.id)).to.be.true
+        // Should be a request to publish
+        expect(eventCID).to.be.null
+        expect(eventNode.isPublished).to.be.false
+        checkAllDone()
+      }
+    )
   })
 
-  it('unsubscribe from the topic', (done) => {
+  it('unsubscribe from the topic', done => {
     const topicB58Str = topicCID.toBaseEncodedString()
-    nodes[1].unsubscribe(topicB58Str, (err) => {
+    nodes[1].unsubscribe(topicB58Str, err => {
       expect(err).to.not.exist
 
       expect(nodes[1].subscriptions.size).to.equal(0)
 
       eventually(() => {
         const topicChildren = nodes[0].me.trees.get(topicB58Str).children
-        expect(topicChildren.find(peer => peer.info.id.isEqual(nodes[1].me.info.id))).to.not.exist
+        expect(
+          topicChildren.find(peer => peer.info.id.isEqual(nodes[1].me.info.id))
+        ).to.not.exist
       }, done)
     })
   })

src/utils/dht-helpers.js

@@ -12,7 +12,7 @@ const TopicNode = require('../../src/dag/topic-node')
 const EventNode = require('../../src/dag/event-node')
 const { eventually, createNodes } = require('../utils')
 
-describe('multiple nodes', function () {
+describe('multiple nodes', function() {
   this.timeout(50000)
   let nodes
   // Topics
@@ -27,72 +27,85 @@ describe('multiple nodes', function () {
   const subscriber = 80
   const subscriberNum = 5
 
-  before((done) => {
+  before(done => {
     createNodes(nodeNumber, (err, p2pNodes) => {
       expect(err).not.to.exist
-      nodes = p2pNodes.map((node) => new Pulsarcast(node))
+      nodes = p2pNodes.map(node => new Pulsarcast(node))
       done()
     })
   })
 
-  it(`starts ${nodeNumber} nodes`, (done) => {
-    each(nodes, (node, cb) => node.start(cb), (err) => {
-      expect(err).to.not.exist
-      nodes.forEach(node => {
-        expect(node.started).to.be.true
-      })
-      done()
-    })
+  it(`starts ${nodeNumber} nodes`, done => {
+    each(
+      nodes,
+      (node, cb) => node.start(cb),
+      err => {
+        expect(err).to.not.exist
+        nodes.forEach(node => {
+          expect(node.started).to.be.true
+        })
+        done()
+      }
+    )
   })
 
-  it('creates a simple parent topic', (done) => {
-    nodes[publisher + 1].createTopic('test-parent', (err, savedCID, topicNode) => {
-      expect(err).to.not.exist
-      expect(topicNode).to.be.an.instanceof(TopicNode)
-      expect(topicNode.subTopics.meta).to.be.an.instanceof(CID)
-      topicNode.getCID((err, cid) => {
+  it('creates a simple parent topic', done => {
+    nodes[publisher + 1].createTopic(
+      'test-parent',
+      (err, savedCID, topicNode) => {
         expect(err).to.not.exist
-        const topicB58Str = cid.toBaseEncodedString()
-        expect(cid.equals(savedCID)).to.be.true
-        // Topic and meta topic
-        expect(nodes[publisher + 1].subscriptions.size).to.equal(2)
-        expect(nodes[publisher + 1].subscriptions.has(topicB58Str)).to.be.true
-        parentTopicCID = cid
-        done()
-      })
-    })
+        expect(topicNode).to.be.an.instanceof(TopicNode)
+        expect(topicNode.subTopics.meta).to.be.an.instanceof(CID)
+        topicNode.getCID((err, cid) => {
+          expect(err).to.not.exist
+          const topicB58Str = cid.toBaseEncodedString()
+          expect(cid.equals(savedCID)).to.be.true
+          // Topic and meta topic
+          expect(nodes[publisher + 1].subscriptions.size).to.equal(2)
+          expect(nodes[publisher + 1].subscriptions.has(topicB58Str)).to.be.true
+          parentTopicCID = cid
+          done()
+        })
+      }
+    )
   })
 
-  it('creates a simple sub-topic', (done) => {
-    nodes[publisher + 1].createTopic('test-subtopic', (err, savedCID, topicNode) => {
-      expect(err).to.not.exist
-      expect(topicNode).to.be.an.instanceof(TopicNode)
-      expect(topicNode.subTopics.meta).to.be.an.instanceof(CID)
-      topicNode.getCID((err, cid) => {
+  it('creates a simple sub-topic', done => {
+    nodes[publisher + 1].createTopic(
+      'test-subtopic',
+      (err, savedCID, topicNode) => {
         expect(err).to.not.exist
-        const topicB58Str = cid.toBaseEncodedString()
-        expect(cid.equals(savedCID)).to.be.true
-        expect(nodes[publisher + 1].subscriptions.size).to.equal(4)
-        expect(nodes[publisher + 1].subscriptions.has(topicB58Str)).to.be.true
-        subTopicCID = cid
-        done()
-      })
-    })
+        expect(topicNode).to.be.an.instanceof(TopicNode)
+        expect(topicNode.subTopics.meta).to.be.an.instanceof(CID)
+        topicNode.getCID((err, cid) => {
+          expect(err).to.not.exist
+          const topicB58Str = cid.toBaseEncodedString()
+          expect(cid.equals(savedCID)).to.be.true
+          expect(nodes[publisher + 1].subscriptions.size).to.equal(4)
+          expect(nodes[publisher + 1].subscriptions.has(topicB58Str)).to.be.true
+          subTopicCID = cid
+          done()
+        })
+      }
+    )
   })
 
-  it('creates a new topic with a parent and sub-topic', (done) => {
-    nodes[publisher].createTopic('test-2.0',
+  it('creates a new topic with a parent and sub-topic', done => {
+    nodes[publisher].createTopic(
+      'test-2.0',
       {
         parent: parentTopicCID.toBaseEncodedString(),
         subTopics: {
           'test-subtopic': subTopicCID.toBaseEncodedString()
         }
-      }, (err, savedCID, topicNode) => {
+      },
+      (err, savedCID, topicNode) => {
         expect(err).to.not.exist
         expect(topicNode).to.be.an.instanceof(TopicNode)
         expect(topicNode.subTopics.meta).to.be.an.instanceof(CID)
         expect(topicNode.parent.equals(parentTopicCID)).to.be.true
-        expect(topicNode.subTopics['test-subtopic'].equals(subTopicCID)).to.be.true
+        expect(topicNode.subTopics['test-subtopic'].equals(subTopicCID)).to.be
+          .true
         topicNode.getCID((err, cid) => {
           expect(err).to.not.exist
           const topicB58Str = cid.toBaseEncodedString()
@@ -103,31 +116,37 @@ describe('multiple nodes', function () {
           topicCID = cid
           done()
         })
-      })
+      }
+    )
   })
 
-  it(`subscribes ${subscriberNum} nodes to the created topic`, (done) => {
+  it(`subscribes ${subscriberNum} nodes to the created topic`, done => {
     const topicB58Str = topicCID.toBaseEncodedString()
     const subscriberNodes = nodes.slice(subscriber, subscriberNum + subscriber)
 
-    each(subscriberNodes, (node, cb) => {
-      node.subscribe(topicB58Str, (err, topicNode) => {
-        expect(err).to.not.exist
-        expect(topicNode).to.be.an.instanceof(TopicNode)
-        expect(node.subscriptions.size).to.equal(1)
-        expect(node.subscriptions.has(topicCID.toBaseEncodedString())).to.be.true
-        expect(topicNode.serialize()).to.deep.equal(topic.serialize())
-
-        // Check tree
-        const {parents} = node.me.trees.get(topicB58Str)
-        expect(parents).to.have.lengthOf.above(0)
-        expect(parents[0].trees.get(topicB58Str).children).to.include(node.me)
-        cb()
-      })
-    }, done)
+    each(
+      subscriberNodes,
+      (node, cb) => {
+        node.subscribe(topicB58Str, (err, topicNode) => {
+          expect(err).to.not.exist
+          expect(topicNode).to.be.an.instanceof(TopicNode)
+          expect(node.subscriptions.size).to.equal(1)
+          expect(node.subscriptions.has(topicCID.toBaseEncodedString())).to.be
+            .true
+          expect(topicNode.serialize()).to.deep.equal(topic.serialize())
+
+          // Check tree
+          const { parents } = node.me.trees.get(topicB58Str)
+          expect(parents).to.have.lengthOf.above(0)
+          expect(parents[0].trees.get(topicB58Str).children).to.include(node.me)
+          cb()
+        })
+      },
+      done
+    )
   })
 
-  it('publishes a message from the non author node', (done) => {
+  it('publishes a message from the non author node', done => {
     const topicB58Str = topicCID.toBaseEncodedString()
     const message = 'foobar'
     // Helper func to run all the expects
@@ -138,7 +157,7 @@ describe('multiple nodes', function () {
     }
     let firstEventNode
     // Event listener
-    const listener = (eventNode) => {
+    const listener = eventNode => {
       // Compare serializes of both events
       if (!firstEventNode) {
         firstEventNode = eventNode
@@ -153,44 +172,53 @@ describe('multiple nodes', function () {
       // Should match subscribed author
       expect(eventNode.author.isEqual(nodes[subscriber].me.info.id)).to.be.true
       // Should match topic author
-      expect(eventNode.publisher.isEqual(nodes[publisher].me.info.id)).to.be.true
+      expect(eventNode.publisher.isEqual(nodes[publisher].me.info.id)).to.be
+        .true
 
       checkAllDone()
     }
     // Setup event listeners
     nodes[subscriber].once(topicB58Str, listener)
     nodes[publisher].once(topicB58Str, listener)
 
-    nodes[subscriber].publish(topicB58Str, message, (err, eventCID, topicNode, eventNode) => {
-      expect(err).to.not.exist
-      expect(eventNode).to.be.an.instanceof(EventNode)
-      expect(topicNode).to.be.an.instanceof(TopicNode)
-      expect(eventNode.topicCID.equals(topicCID)).to.be.true
-      expect(eventNode.payload.toString()).to.be.equal(message)
-      expect(eventNode.author.isEqual(nodes[subscriber].me.info.id)).to.be.true
-      // Should be a request to publish
-      expect(eventCID).to.be.null
-      expect(eventNode.isPublished).to.be.false
-
-      checkAllDone()
-    })
+    nodes[subscriber].publish(
+      topicB58Str,
+      message,
+      (err, eventCID, topicNode, eventNode) => {
+        expect(err).to.not.exist
+        expect(eventNode).to.be.an.instanceof(EventNode)
+        expect(topicNode).to.be.an.instanceof(TopicNode)
+        expect(eventNode.topicCID.equals(topicCID)).to.be.true
+        expect(eventNode.payload.toString()).to.be.equal(message)
+        expect(eventNode.author.isEqual(nodes[subscriber].me.info.id)).to.be
+          .true
+        // Should be a request to publish
+        expect(eventCID).to.be.null
+        expect(eventNode.isPublished).to.be.false
+
+        checkAllDone()
+      }
+    )
   })
 
-  it('unsubscribe from the topic', (done) => {
+  it('unsubscribe from the topic', done => {
     const topicB58Str = topicCID.toBaseEncodedString()
-    nodes[subscriber].unsubscribe(topicB58Str, (err) => {
+    nodes[subscriber].unsubscribe(topicB58Str, err => {
       expect(err).to.not.exist
 
       expect(nodes[subscriber].subscriptions.size).to.equal(0)
 
       eventually(() => {
-        const topicChildren = nodes[publisher].me.trees.get(topicB58Str).children
-        expect(topicChildren.find(peer => peer.info.id.isEqual(nodes[1].me.info.id))).to.not.exist
+        const topicChildren = nodes[publisher].me.trees.get(topicB58Str)
+          .children
+        expect(
+          topicChildren.find(peer => peer.info.id.isEqual(nodes[1].me.info.id))
+        ).to.not.exist
       }, done)
     })
   })
 
-  it('dissemination tress are cleaned up on connection close', (done) => {
+  it('dissemination tress are cleaned up on connection close', done => {
     const topicB58Str = topicCID.toBaseEncodedString()
     const droppingNode = nodes[subscriber + 1]
     const droppingNodeId = droppingNode.me.info.id.toB58String()
@@ -211,10 +239,14 @@ describe('multiple nodes', function () {
       eventually(() => {
         // Check dropping node is not present in any tree
         childrenNodes.forEach(child => {
-          expect(child.me.trees.get(topicB58Str).parents).to.not.include(child.peers.get(droppingNodeId))
+          expect(child.me.trees.get(topicB58Str).parents).to.not.include(
+            child.peers.get(droppingNodeId)
+          )
         })
         parentNodes.forEach(parent => {
-          expect(parent.me.trees.get(topicB58Str).children).to.not.include(parent.peers.get(droppingNodeId))
+          expect(parent.me.trees.get(topicB58Str).children).to.not.include(
+            parent.peers.get(droppingNodeId)
+          )
         })
       }, done)
     })

test/integration/2-nodes.js

@@ -7,7 +7,7 @@ const DHT = require('libp2p-kad-dht')
 const libp2p = require('libp2p')
 
 class Node extends libp2p {
-  constructor ({ peerInfo, peerBook }) {
+  constructor({ peerInfo, peerBook }) {
     const modules = {
       transport: [TCP],
       streamMuxer: [spdy],

test/integration/multiple-nodes.js

@@ -6,19 +6,22 @@ const { waterfall, eachOfLimit, mapLimit } = require('async')
 
 const Node = require('./test-node')
 
-function createNode (maddr, callback) {
-  waterfall([
-    (cb) => PeerId.create({ bits: 1024 }, cb),
-    (id, cb) => PeerInfo.create(id, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add(maddr)
-      cb(null, new Node({ peerInfo }))
-    },
-    (node, cb) => node.start((err) => cb(err, node))
-  ], callback)
+function createNode(maddr, callback) {
+  waterfall(
+    [
+      cb => PeerId.create({ bits: 1024 }, cb),
+      (id, cb) => PeerInfo.create(id, cb),
+      (peerInfo, cb) => {
+        peerInfo.multiaddrs.add(maddr)
+        cb(null, new Node({ peerInfo }))
+      },
+      (node, cb) => node.start(err => cb(err, node))
+    ],
+    callback
+  )
 }
 
-function createNodes (nodeNumber, callback) {
+function createNodes(nodeNumber, callback) {
   const maddrs = []
   for (let i = 0; i < nodeNumber; i++) {
     maddrs.push('/ip4/127.0.0.1/tcp/0')
@@ -27,18 +30,23 @@ function createNodes (nodeNumber, callback) {
     if (err) return callback(err)
 
     // Connect nodes sequentially, essentially creating a ring
-    eachOfLimit(nodes, 10, (node, index, cb) => {
-      let nextNode = nodes[index + 1]
-      // End of node list
-      if (!nextNode) return cb()
-      node.dial(nextNode.peerInfo, cb)
-    }, (err) => {
-      callback(err, nodes)
-    })
+    eachOfLimit(
+      nodes,
+      10,
+      (node, index, cb) => {
+        let nextNode = nodes[index + 1]
+        // End of node list
+        if (!nextNode) return cb()
+        node.dial(nextNode.peerInfo, cb)
+      },
+      err => {
+        callback(err, nodes)
+      }
+    )
   })
 }
 
-function eventually (func, callback) {
+function eventually(func, callback) {
   const polling = setInterval(() => {
     try {
       func()