chore: address review

src/pubsub/index.js

@@ -402,10 +402,12 @@ class PubsubBaseProtocol extends EventEmitter {
       return
     }
 
+    const normalizedMessage = utils.normalizeOutRpcMessage(msg)
+
     // Emit to self
-    this._emitMessage(msg)
+    this._emitMessage(normalizedMessage)
 
-    this._publishFrom(msg)
+    this._publish(normalizedMessage)
   }
 
   /**
@@ -430,16 +432,6 @@ class PubsubBaseProtocol extends EventEmitter {
     return utils.msgId(msg.from, msg.seqno)
   }
 
-  /**
-   * Process publish message received from a peer
-   * @abstract
-   * @param {InMessage} msg
-   * @returns {void}
-   */
-  _publishFrom (msg) {
-    throw errcode(new Error('_publishFrom must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
-  }
-
   /**
    * Whether to accept a message from a peer
    * Override to create a graylist
@@ -452,25 +444,23 @@ class PubsubBaseProtocol extends EventEmitter {
   }
 
   /**
-   * Overriding the implementation of _decodeRpc should use the appropriate router protobuf.
    * Decode Uint8Array into an RPC object.
-   * @abstract
+   * This can be override to use a custom router protobuf.
    * @param {Uint8Array} bytes
    * @returns {RPC}
    */
   _decodeRpc (bytes) {
-    throw errcode(new Error('_decodeRpc must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
+    return message.rpc.RPC.decode(bytes)
   }
 
   /**
-   * Overriding the implementation of _encodeRpc should use the appropriate router protobuf.
    * Encode RPC object into a Uint8Array.
-   * @abstract
+   * This can be override to use a custom router protobuf.
    * @param {RPC} rpc
    * @returns {Uint8Array}
    */
   _encodeRpc (rpc) {
-    throw errcode(new Error('_encodeRpc must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
+    return message.rpc.RPC.encode(rpc)
   }
 
   /**
@@ -581,14 +571,18 @@ class PubsubBaseProtocol extends EventEmitter {
     this.log('publish', topics, message)
 
     const from = this.peerId.toB58String()
-    const msgObject = {
+    let msgObject = {
       receivedFrom: from,
       from: from,
       data: message,
       seqno: utils.randomSeqno(),
       topicIDs: topics
     }
 
+    // ensure that any operations performed on the message will include the signature
+    const outMsg = await this._buildMessage(msgObject)
+    msgObject = utils.normalizeInRpcMessage(outMsg)
+
     // Emit to self if I'm interested and emitSelf enabled
     this.emitSelf && this._emitMessage(msgObject)