feat: add support for signing

package.json

@@ -61,7 +61,7 @@
     "debug": "^4.1.1",
     "length-prefixed-stream": "^2.0.0",
     "libp2p-crypto": "~0.16.1",
-    "libp2p-pubsub": "~0.0.4",
+    "libp2p-pubsub": "~0.1.0",
     "protons": "^1.0.1",
     "pull-length-prefixed": "^1.3.2",
     "pull-pushable": "^2.2.0",

src/index.js

@@ -11,6 +11,8 @@ const config = require('./config')
 const multicodec = config.multicodec
 const ensureArray = utils.ensureArray
 const setImmediate = require('async/setImmediate')
+const asyncMap = require('async/map')
+const noop = () => {}
 
 /**
  * FloodSub (aka dumbsub is an implementation of pubsub focused on
@@ -158,11 +160,13 @@ class FloodSub extends BaseProtocol {
    * @override
    * @param {Array<string>|string} topics
    * @param {Array<any>|any} messages
+   * @param {function(Error)} callback
    * @returns {undefined}
    *
    */
-  publish (topics, messages) {
+  publish (topics, messages, callback) {
     assert(this.started, 'FloodSub is not started')
+    callback = callback || noop
 
     this.log('publish', topics, messages)
 
@@ -171,25 +175,29 @@ class FloodSub extends BaseProtocol {
 
     const from = this.libp2p.peerInfo.id.toB58String()
 
-    const buildMessage = (msg) => {
+    const buildMessage = (msg, cb) => {
       const seqno = utils.randomSeqno()
       this.seenCache.put(utils.msgId(from, seqno))
 
-      return {
+      this._buildMessage({
         from: from,
         data: msg,
         seqno: seqno,
         topicIDs: topics
-      }
+      }, cb)
     }
 
-    const msgObjects = messages.map(buildMessage)
+    asyncMap(messages, buildMessage, (err, msgObjects) => {
+      if (err) return callback(err)
+
+      // Emit to self if I'm interested
+      this._emitMessages(topics, msgObjects)
 
-    // Emit to self if I'm interested
-    this._emitMessages(topics, msgObjects)
+      // send to all the other peers
+      this._forwardMessages(topics, msgObjects)
 
-    // send to all the other peers
-    this._forwardMessages(topics, msgObjects)
+      callback(null)
+    })
   }
 
   /**