chore: address review

src/pubsub/index.js

@@ -216,7 +216,7 @@ class PubsubBaseProtocol extends EventEmitter {
    * @private
    * @param {Object} props
    * @param {string} props.protocol
-   * @param {DuplexIterableStream} props.stream
+   * @param {MuxedStream} props.stream
    * @param {Connection} props.connection - connection
    */
   _onIncomingStream ({ protocol, stream, connection }) {
@@ -225,8 +225,7 @@ class PubsubBaseProtocol extends EventEmitter {
     const peer = this._addPeer(peerId, protocol)
     peer.attachInboundStream(stream)
 
-    // @ts-ignore - peer.inboundStream maybe null
-    this._processMessages(idB58Str, peer.inboundStream, peer)
+    peer.inboundStream && this._processMessages(idB58Str, peer.inboundStream, peer)
   }
 
   /**
@@ -243,7 +242,6 @@ class PubsubBaseProtocol extends EventEmitter {
     try {
       const { stream, protocol } = await conn.newStream(this.multicodecs)
       const peer = this._addPeer(peerId, protocol)
-      // @ts-ignore MuxedStream is not DuplexIterableStream
       await peer.attachOutboundStream(stream)
     } catch (err) {
       this.log.err(err)
@@ -333,7 +331,7 @@ class PubsubBaseProtocol extends EventEmitter {
    * Responsible for processing each RPC message received by other peers.
    *
    * @param {string} idB58Str - peer id string in base58
-   * @param {DuplexIterableStream} stream - inbound stream
+   * @param {MuxedStream} stream - inbound stream
    * @param {PeerStreams} peerStreams - PubSub peer
    * @returns {Promise<void>}
    */
@@ -342,8 +340,8 @@ class PubsubBaseProtocol extends EventEmitter {
       await pipe(
         stream,
         async (source) => {
-          // @ts-ignore - DuplexIterableStream isn't defined as iterable
           for await (const data of source) {
+            // @ts-ignore data slice from BufferList
             const rpcBytes = data instanceof Uint8Array ? data : data.slice()
             const rpcMsg = this._decodeRpc(rpcBytes)
 
@@ -732,7 +730,8 @@ class PubsubBaseProtocol extends EventEmitter {
 
 /**
  * @typedef {any} Libp2p
- * @typedef {import('./peer-streams').DuplexIterableStream} DuplexIterableStream
+ * @typedef {object} MuxedStream
+ * @type import('../stream-muxer/types').MuxedStream
  * @typedef {import('../connection/connection')} Connection
  * @typedef {import('./message').RPC} RPC
  * @typedef {import('./message').SubOpts} RPCSubOpts
@@ -740,7 +739,8 @@ class PubsubBaseProtocol extends EventEmitter {
  * @typedef {import('./signature-policy').SignaturePolicyType} SignaturePolicyType
  */
 
+PubsubBaseProtocol.message = message
+PubsubBaseProtocol.utils = utils
+PubsubBaseProtocol.SignaturePolicy = SignaturePolicy
+
 module.exports = PubsubBaseProtocol
-module.exports.message = message
-module.exports.utils = utils
-module.exports.SignaturePolicy = SignaturePolicy

src/pubsub/message/sign.js

@@ -40,11 +40,12 @@ async function verifySignature (message) {
   const baseMessage = { ...message }
   delete baseMessage.signature
   delete baseMessage.key
-  // @ts-ignore - from is optional
-  baseMessage.from = PeerId.createFromCID(baseMessage.from).toBytes()
+
   const bytes = uint8ArrayConcat([
     SignPrefix,
-    Message.encode(baseMessage)
+    Message.encode(Object.assign(baseMessage, {
+      from: baseMessage.from && PeerId.createFromCID(baseMessage.from).toBytes()
+    }))
   ])
 
   // Get the public key
@@ -64,14 +65,14 @@ async function verifySignature (message) {
  */
 async function messagePublicKey (message) {
   // should be available in the from property of the message (peer id)
-  // @ts-ignore - from is optional
+  // @ts-ignore - from type changed
   const from = PeerId.createFromCID(message.from)
 
   if (message.key) {
     const keyPeerId = await PeerId.createFromPubKey(message.key)
 
     // the key belongs to the sender, return the key
-    if (keyPeerId.isEqual(from)) return keyPeerId.pubKey
+    if (keyPeerId.equals(from)) return keyPeerId.pubKey
     // We couldn't validate pubkey is from the originator, error
     throw new Error('Public Key does not match the originator')
   } else if (from.pubKey) {

src/pubsub/peer-streams.js

@@ -20,12 +20,14 @@ log.error = debug('libp2p-pubsub:peer-streams:error')
  * @param {Uint8Array} source
  * @returns {Promise<Uint8Array>}
  *
- * @typedef {object} DuplexIterableStream
- * @property {Sink} sink
- * @property {AsyncIterator<Uint8Array>} source
+ * @typedef {object} MuxedStream
+ * @type import('../stream-muxer/types').MuxedStream
  *
  * @typedef PeerId
  * @type import('peer-id')
+ *
+ * @typedef PushableStream
+ * @type import('it-pushable').Pushable<Uint8Array>
  */
 
 /**
@@ -54,33 +56,33 @@ class PeerStreams extends EventEmitter {
      * The raw outbound stream, as retrieved from conn.newStream
      *
      * @private
-     * @type {null|DuplexIterableStream}
+     * @type {null|MuxedStream}
      */
     this._rawOutboundStream = null
     /**
      * The raw inbound stream, as retrieved from the callback from libp2p.handle
      *
      * @private
-     * @type {null|DuplexIterableStream}
+     * @type {null|MuxedStream}
      */
     this._rawInboundStream = null
     /**
      * An AbortController for controlled shutdown of the inbound stream
      *
      * @private
-     * @type {null|AbortController}
+     * @type {AbortController}
      */
-    this._inboundAbortController = null
+    this._inboundAbortController = new AbortController()
     /**
      * Write stream -- its preferable to use the write method
      *
-     * @type {null|import('it-pushable').Pushable<Uint8Array>}
+     * @type {null|PushableStream}
      */
     this.outboundStream = null
     /**
      * Read stream
      *
-     * @type {null|DuplexIterableStream}
+     * @type {null|MuxedStream}
      */
     this.inboundStream = null
   }
@@ -123,7 +125,7 @@ class PeerStreams extends EventEmitter {
   /**
    * Attach a raw inbound stream and setup a read stream
    *
-   * @param {DuplexIterableStream} stream
+   * @param {MuxedStream} stream
    * @returns {void}
    */
   attachInboundStream (stream) {
@@ -131,15 +133,13 @@ class PeerStreams extends EventEmitter {
     // The inbound stream is:
     // - abortable, set to only return on abort, rather than throw
     // - transformed with length-prefix transform
-    this._inboundAbortController = new AbortController()
     this._rawInboundStream = stream
-    // @ts-ignore - abortable returns AsyncIterable and not a DuplexIterableStream
+    // @ts-ignore - abortable returns AsyncIterable and not a MuxedStream
     this.inboundStream = abortable(
       pipe(
         this._rawInboundStream,
         lp.decode()
       ),
-      // @ts-ignore - possibly null
       this._inboundAbortController.signal,
       { returnOnAbort: true }
     )
@@ -150,30 +150,26 @@ class PeerStreams extends EventEmitter {
   /**
    * Attach a raw outbound stream and setup a write stream
    *
-   * @param {DuplexIterableStream} stream
+   * @param {MuxedStream} stream
    * @returns {Promise<void>}
    */
   async attachOutboundStream (stream) {
     // If an outbound stream already exists,
     // gently close it
     const _prevStream = this.outboundStream
-    if (_prevStream) {
+    if (this.outboundStream) {
       // End the stream without emitting a close event
-      // @ts-ignore - outboundStream may be null
-      await this.outboundStream.end(false)
+      await this.outboundStream.end()
     }
 
     this._rawOutboundStream = stream
     this.outboundStream = pushable({
       onEnd: (shouldEmit) => {
         // close writable side of the stream
-        // @ts-ignore - DuplexIterableStream does not define reset
         this._rawOutboundStream && this._rawOutboundStream.reset && this._rawOutboundStream.reset()
         this._rawOutboundStream = null
         this.outboundStream = null
-        // @ts-ignore - shouldEmit is `Error | undefined` so condition is
-        // always false
-        if (shouldEmit !== false) {
+        if (shouldEmit) {
           this.emit('close')
         }
       }
@@ -205,7 +201,6 @@ class PeerStreams extends EventEmitter {
     }
     // End the inbound stream
     if (this.inboundStream) {
-      // @ts-ignore - possibly null
       this._inboundAbortController.abort()
     }