fix: _emitMessages should not emit normalized messages (#79)

Author: jacobheun

package.json

@@ -53,7 +53,8 @@
     "libp2p-tcp": "~0.13.0",
     "lodash": "^4.17.11",
     "peer-id": "~0.12.2",
-    "peer-info": "~0.15.1"
+    "peer-info": "~0.15.1",
+    "sinon": "^7.3.2"
   },
   "dependencies": {
     "async": "^2.6.2",

src/index.js

@@ -179,20 +179,22 @@ class FloodSub extends BaseProtocol {
       const seqno = utils.randomSeqno()
       this.seenCache.put(utils.msgId(from, seqno))
 
-      this._buildMessage({
+      const message = {
         from: from,
         data: msg,
         seqno: seqno,
         topicIDs: topics
-      }, cb)
+      }
+
+      // Emit to self if I'm interested
+      this._emitMessages(topics, [message])
+
+      this._buildMessage(message, cb)
     }
 
     asyncMap(messages, buildMessage, (err, msgObjects) => {
       if (err) return callback(err)
 
-      // Emit to self if I'm interested
-      this._emitMessages(topics, msgObjects)
-
       // send to all the other peers
       this._forwardMessages(topics, msgObjects)
 

test/pubsub.spec.js

@@ -0,0 +1,89 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 5] */
+'use strict'
+
+const chai = require('chai')
+chai.use(require('dirty-chai'))
+const expect = chai.expect
+const sinon = require('sinon')
+
+const Floodsub = require('../src')
+const { createNode } = require('./utils')
+const { utils } = require('libp2p-pubsub')
+
+describe('pubsub', () => {
+  let floodsub
+  let libp2p
+
+  before((done) => {
+    createNode('/ip4/127.0.0.1/tcp/0', (err, node) => {
+      expect(err).to.not.exist()
+      libp2p = node
+      floodsub = new Floodsub(libp2p)
+      done(err)
+    })
+  })
+
+  beforeEach(done => {
+    floodsub.start(done)
+  })
+
+  afterEach(done => {
+    sinon.restore()
+    floodsub.stop(done)
+  })
+
+  describe('publish', () => {
+    it('should emit non normalized messages', (done) => {
+      sinon.spy(floodsub, '_emitMessages')
+      sinon.spy(utils, 'randomSeqno')
+
+      const topic = 'my-topic'
+      const message = Buffer.from('a neat message')
+
+      floodsub.publish(topic, message, (err) => {
+        expect(err).to.not.exist()
+        expect(floodsub._emitMessages.callCount).to.eql(1)
+
+        const [topics, messages] = floodsub._emitMessages.getCall(0).args
+        expect(topics).to.eql([topic])
+        expect(messages).to.eql([{
+          from: libp2p.peerInfo.id.toB58String(),
+          data: message,
+          seqno: utils.randomSeqno.getCall(0).returnValue,
+          topicIDs: topics
+        }])
+        done()
+      })
+    })
+
+    it('should forward normalized messages', (done) => {
+      sinon.spy(floodsub, '_forwardMessages')
+      sinon.spy(utils, 'randomSeqno')
+
+      const topic = 'my-topic'
+      const message = Buffer.from('a neat message')
+
+      floodsub.publish(topic, message, (err) => {
+        expect(err).to.not.exist()
+        expect(floodsub._forwardMessages.callCount).to.eql(1)
+        const [topics, messages] = floodsub._forwardMessages.getCall(0).args
+
+        floodsub._buildMessage({
+          from: libp2p.peerInfo.id.toB58String(),
+          data: message,
+          seqno: utils.randomSeqno.getCall(0).returnValue,
+          topicIDs: topics
+        }, (err, expected) => {
+          expect(err).to.not.exist()
+
+          expect(topics).to.eql([topic])
+          expect(messages).to.eql([
+            expected
+          ])
+          done()
+        })
+      })
+    })
+  })
+})