@@ -11,7 +11,7 @@ const { pipe } = require('it-pipe')
1111const MulticodecTopology = require ( '../topology/multicodec-topology' )
1212const { codes } = require ( './errors' )
1313
14- const message = require ( './message' )
14+ const { RPC } = require ( './message/rpc ' )
1515const PeerStreams = require ( './peer-streams' )
1616const { SignaturePolicy } = require ( './signature-policy' )
1717const utils = require ( './utils' )
@@ -27,10 +27,10 @@ const {
2727 * @typedef {import('bl') } BufferList
2828 * @typedef {import('../stream-muxer/types').MuxedStream } MuxedStream
2929 * @typedef {import('../connection/connection') } Connection
30- * @typedef {import('./message/types').RPC } RPC
31- * @typedef {import('./message/types').SubOpts } RPCSubOpts
32- * @typedef {import('./message/types').Message } RPCMessage
3330 * @typedef {import('./signature-policy').SignaturePolicyType } SignaturePolicyType
31+ * @typedef {import('./message/rpc').IRPC } IRPC
32+ * @typedef {import('./message/rpc').RPC.SubOpts } RPCSubOpts
33+ * @typedef {import('./message/rpc').RPC.Message } RPCMessage
3434 */
3535
3636/**
@@ -382,7 +382,7 @@ class PubsubBaseProtocol extends EventEmitter {
382382
383383 if ( subs . length ) {
384384 // update peer subscriptions
385- subs . forEach ( ( /** @type { RPCSubOpts } */ subOpt ) => {
385+ subs . forEach ( ( subOpt ) => {
386386 this . _processRpcSubOpt ( idB58Str , subOpt )
387387 } )
388388 this . emit ( 'pubsub:subscription-change' , peerStreams . id , subs )
@@ -396,7 +396,7 @@ class PubsubBaseProtocol extends EventEmitter {
396396 if ( msgs . length ) {
397397 // @ts -ignore RPC message is modified
398398 msgs . forEach ( ( message ) => {
399- if ( ! ( this . canRelayMessage || message . topicIDs . some ( ( /** @type { string } */ topic ) => this . subscriptions . has ( topic ) ) ) ) {
399+ if ( ! ( this . canRelayMessage || ( message . topicIDs && message . topicIDs . some ( ( topic ) => this . subscriptions . has ( topic ) ) ) ) ) {
400400 this . log ( 'received message we didn\'t subscribe to. Dropping.' )
401401 return
402402 }
@@ -411,11 +411,15 @@ class PubsubBaseProtocol extends EventEmitter {
411411 * Handles a subscription change from a peer
412412 *
413413 * @param {string } id
414- * @param {RPCSubOpts } subOpt
414+ * @param {RPC.ISubOpts } subOpt
415415 */
416416 _processRpcSubOpt ( id , subOpt ) {
417417 const t = subOpt . topicID
418418
419+ if ( ! t ) {
420+ return
421+ }
422+
419423 let topicSet = this . topics . get ( t )
420424 if ( ! topicSet ) {
421425 topicSet = new Set ( )
@@ -473,13 +477,14 @@ class PubsubBaseProtocol extends EventEmitter {
473477 * The default msgID implementation
474478 * Child class can override this.
475479 *
476- * @param {RPCMessage } msg - the message object
480+ * @param {InMessage } msg - the message object
477481 * @returns {Uint8Array } message id as bytes
478482 */
479483 getMsgId ( msg ) {
480484 const signaturePolicy = this . globalSignaturePolicy
481485 switch ( signaturePolicy ) {
482486 case SignaturePolicy . StrictSign :
487+ // @ts -ignore seqno is optional in protobuf definition but it will exist
483488 return utils . msgId ( msg . from , msg . seqno )
484489 case SignaturePolicy . StrictNoSign :
485490 return utils . noSignMsgId ( msg . data )
@@ -508,25 +513,25 @@ class PubsubBaseProtocol extends EventEmitter {
508513 * @returns {RPC }
509514 */
510515 _decodeRpc ( bytes ) {
511- return message . rpc . RPC . decode ( bytes )
516+ return RPC . decode ( bytes )
512517 }
513518
514519 /**
515520 * Encode RPC object into a Uint8Array.
516521 * This can be override to use a custom router protobuf.
517522 *
518- * @param {RPC } rpc
523+ * @param {IRPC } rpc
519524 * @returns {Uint8Array }
520525 */
521526 _encodeRpc ( rpc ) {
522- return message . rpc . RPC . encode ( rpc )
527+ return RPC . encode ( rpc ) . finish ( )
523528 }
524529
525530 /**
526531 * Send an rpc object to a peer
527532 *
528533 * @param {string } id - peer id
529- * @param {RPC } rpc
534+ * @param {IRPC } rpc
530535 * @returns {void }
531536 */
532537 _sendRpc ( id , rpc ) {
@@ -592,12 +597,12 @@ class PubsubBaseProtocol extends EventEmitter {
592597 default :
593598 throw errcode ( new Error ( 'Cannot validate message: unhandled signature policy: ' + signaturePolicy ) , codes . ERR_UNHANDLED_SIGNATURE_POLICY )
594599 }
600+
595601 for ( const topic of message . topicIDs ) {
596602 const validatorFn = this . topicValidators . get ( topic )
597- if ( ! validatorFn ) {
598- continue // eslint-disable-line
603+ if ( validatorFn ) {
604+ await validatorFn ( topic , message )
599605 }
600- await validatorFn ( topic , message )
601606 }
602607 }
603608
@@ -606,8 +611,8 @@ class PubsubBaseProtocol extends EventEmitter {
606611 * Should be used by the routers to create the message to send.
607612 *
608613 * @protected
609- * @param {RPCMessage } message
610- * @returns {Promise<RPCMessage > }
614+ * @param {InMessage } message
615+ * @returns {Promise<InMessage > }
611616 */
612617 _buildMessage ( message ) {
613618 const signaturePolicy = this . globalSignaturePolicy
@@ -617,7 +622,7 @@ class PubsubBaseProtocol extends EventEmitter {
617622 message . seqno = utils . randomSeqno ( )
618623 return signMessage ( this . peerId , utils . normalizeOutRpcMessage ( message ) )
619624 case SignaturePolicy . StrictNoSign :
620- return message
625+ return Promise . resolve ( message )
621626 default :
622627 throw errcode ( new Error ( 'Cannot build message: unhandled signature policy: ' + signaturePolicy ) , codes . ERR_UNHANDLED_SIGNATURE_POLICY )
623628 }
@@ -663,29 +668,30 @@ class PubsubBaseProtocol extends EventEmitter {
663668 this . log ( 'publish' , topic , message )
664669
665670 const from = this . peerId . toB58String ( )
666- let msgObject = {
671+ const msgObject = {
667672 receivedFrom : from ,
668673 data : message ,
669674 topicIDs : [ topic ]
670675 }
671676
672677 // ensure that the message follows the signature policy
673678 const outMsg = await this . _buildMessage ( msgObject )
674- msgObject = utils . normalizeInRpcMessage ( outMsg )
679+ // @ts -ignore different type as from is converted
680+ const msg = utils . normalizeInRpcMessage ( outMsg )
675681
676682 // Emit to self if I'm interested and emitSelf enabled
677- this . emitSelf && this . _emitMessage ( msgObject )
683+ this . emitSelf && this . _emitMessage ( msg )
678684
679685 // send to all the other peers
680- await this . _publish ( msgObject )
686+ await this . _publish ( msg )
681687 }
682688
683689 /**
684690 * Overriding the implementation of publish should handle the appropriate algorithms for the publish/subscriber implementation.
685691 * For example, a Floodsub implementation might simply publish each message to each topic for every peer
686692 *
687693 * @abstract
688- * @param {InMessage } message
694+ * @param {InMessage|RPCMessage } message
689695 * @returns {Promise<void> }
690696 *
691697 */
@@ -744,7 +750,6 @@ class PubsubBaseProtocol extends EventEmitter {
744750 }
745751}
746752
747- PubsubBaseProtocol . message = message
748753PubsubBaseProtocol . utils = utils
749754PubsubBaseProtocol . SignaturePolicy = SignaturePolicy
750755
0 commit comments