Various floodsub issues

src/index.js

@@ -62,14 +62,14 @@ class FloodSub extends EventEmitter {
   _dialPeer (peerInfo, callback) {
     callback = callback || function noop () {}
     const idB58Str = peerInfo.id.toB58String()
-    log('dialing %s', idB58Str)
 
     // If already have a PubSub conn, ignore
     const peer = this.peers.get(idB58Str)
     if (peer && peer.isConnected) {
       return setImmediate(() => callback())
     }
 
+    log('dialing %s', idB58Str)
     this.libp2p.dial(peerInfo, multicodec, (err, conn) => {
       if (err) {
         log.err(err)
@@ -82,6 +82,7 @@ class FloodSub extends EventEmitter {
 
   _onDial (peerInfo, conn, callback) {
     const idB58Str = peerInfo.id.toB58String()
+    log('connected', idB58Str)
 
     // If already had a dial to me, just add the conn
     if (!this.peers.has(idB58Str)) {
@@ -121,7 +122,7 @@ class FloodSub extends EventEmitter {
       pull.map((data) => pb.rpc.RPC.decode(data)),
       pull.drain(
         (rpc) => this._onRpc(idB58Str, rpc),
-        (err) => this._onConnectionEnd(idB58Str, err)
+        (err) => this._onConnectionEnd(idB58Str, conn, err)
       )
     )
   }
@@ -131,6 +132,7 @@ class FloodSub extends EventEmitter {
       return
     }
 
+    log('rpc from', idB58Str, rpc)
     const subs = rpc.subscriptions
     const msgs = rpc.msgs
 
@@ -157,20 +159,25 @@ class FloodSub extends EventEmitter {
       this.cache.put(seqno)
 
       // 2. emit to self
-      this._emitMessages(msg.topicIDs, [msg])
+      this._emitMessages(msg.topicCIDs, [msg])
 
       // 3. propagate msg to others
-      this._forwardMessages(msg.topicIDs, [msg])
+      this._forwardMessages(msg.topicCIDs, [msg])
     })
   }
 
-  _onConnectionEnd (idB58Str, err) {
+  _onConnectionEnd (idB58Str, conn, err) {
     // socket hang up, means the one side canceled
     if (err && err.message !== 'socket hang up') {
       log.err(err)
     }
 
-    this.peers.delete(idB58Str)
+    log('connection ended', idB58Str)
+    const peer = this.peers.get(idB58Str)
+    if (peer && peer.conn === conn) {
+      log('delete peer', idB58Str)
+      this.peers.delete(idB58Str)
+    }
   }
 
   _emitMessages (topics, messages) {
@@ -180,7 +187,14 @@ class FloodSub extends EventEmitter {
       }
 
       messages.forEach((message) => {
-        this.emit(topic, message)
+        // Convert RPC message to API message
+        const m = {
+          from: message.from,
+          data: message.data,
+          seqno: message.seqno,
+          topicIDs: message.topicCIDs
+        }
+        this.emit(topic, m)
       })
     })
   }
@@ -277,7 +291,7 @@ class FloodSub extends EventEmitter {
         from: from,
         data: msg,
         seqno: new Buffer(seqno),
-        topicIDs: topics
+        topicCIDs: topics
       }
     }
 
@@ -287,7 +301,7 @@ class FloodSub extends EventEmitter {
     this._emitMessages(topics, msgObjects)
 
     // send to all the other peers
-    this._forwardMessages(topics, messages.map(buildMessage))
+    this._forwardMessages(topics, msgObjects)
   }
 
   /**
@@ -321,7 +335,11 @@ class FloodSub extends EventEmitter {
    * @returns {undefined}
    */
   unsubscribe (topics) {
-    assert(this.started, 'FloodSub is not started')
+    // Avoid race conditions, by quietly ignoring unsub when shutdown.
+    if (!this.started) {
+      return
+    }
+
     topics = ensureArray(topics)
 
     topics.forEach((topic) => this.subscriptions.delete(topic))

src/message/rpc.proto.js

@@ -13,6 +13,6 @@ message RPC {
     optional string from = 1;
     optional bytes data = 2;
     optional bytes seqno = 3;
-    repeated string topicIDs = 4; // CID of topic descriptor object
+    repeated string topicCIDs = 4; // CID of topic descriptor object
   }
 }`

src/peer.js

@@ -155,9 +155,6 @@ class Peer {
    * @returns {undefined}
    */
   close (callback) {
-    if (!this.conn || !this.stream) {
-      // no connection to close
-    }
     // end the pushable pull-stream
     if (this.stream) {
       this.stream.end()