This repository has been archived by the owner on Jun 26, 2023. It is now read-only.
chore: update pubsub interface, multiaddr and remove protons #89
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,7 +11,7 @@ const { pipe } = require('it-pipe') | |
| const MulticodecTopology = require('../topology/multicodec-topology') | ||
| const { codes } = require('./errors') | ||
|
|
||
| const { RPC } = require('./message/rpc') | ||
| const PeerStreams = require('./peer-streams') | ||
| const { SignaturePolicy } = require('./signature-policy') | ||
| const utils = require('./utils') | ||
|
|
@@ -27,10 +27,10 @@ const { | |
| * @typedef {import('bl')} BufferList | ||
| * @typedef {import('../stream-muxer/types').MuxedStream} MuxedStream | ||
| * @typedef {import('../connection/connection')} Connection | ||
| * @typedef {import('./signature-policy').SignaturePolicyType} SignaturePolicyType | ||
| * @typedef {import('./message/rpc').IRPC} IRPC | ||
| * @typedef {import('./message/rpc').RPC.SubOpts} RPCSubOpts | ||
| * @typedef {import('./message/rpc').RPC.Message} RPCMessage | ||
| */ | ||
|
|
||
| /** | ||
|
|
@@ -382,7 +382,7 @@ class PubsubBaseProtocol extends EventEmitter { | |
|
|
||
| if (subs.length) { | ||
| // update peer subscriptions | ||
| subs.forEach((subOpt) => { | ||
| this._processRpcSubOpt(idB58Str, subOpt) | ||
| }) | ||
| this.emit('pubsub:subscription-change', peerStreams.id, subs) | ||
|
|
@@ -396,7 +396,7 @@ class PubsubBaseProtocol extends EventEmitter { | |
| if (msgs.length) { | ||
| // @ts-ignore RPC message is modified | ||
| msgs.forEach((message) => { | ||
| if (!(this.canRelayMessage || (message.topicIDs && message.topicIDs.some((topic) => this.subscriptions.has(topic))))) { | ||
|
There was a problem hiding this comment. oo, was this a bug that types picked up? There was a problem hiding this comment. This is because the protobuf definition does not state topicIDs as required. As far as I know, all implementations send an empty arrau and not undefined and this has not been a problem. But could be as it is not spec'ed as required |
||
| this.log('received message we didn\'t subscribe to. Dropping.') | ||
| return | ||
| } | ||
|
|
@@ -411,11 +411,15 @@ class PubsubBaseProtocol extends EventEmitter { | |
| * Handles a subscription change from a peer | ||
| * | ||
| * @param {string} id | ||
| * @param {RPC.ISubOpts} subOpt | ||
| */ | ||
| _processRpcSubOpt (id, subOpt) { | ||
| const t = subOpt.topicID | ||
|
|
||
| if (!t) { | ||
| return | ||
| } | ||
|
|
||
| let topicSet = this.topics.get(t) | ||
| if (!topicSet) { | ||
| topicSet = new Set() | ||
|
|
@@ -473,13 +477,14 @@ class PubsubBaseProtocol extends EventEmitter { | |
| * The default msgID implementation | ||
| * Child class can override this. | ||
| * | ||
| * @param {InMessage} msg - the message object | ||
| * @returns {Uint8Array} message id as bytes | ||
| */ | ||
| getMsgId (msg) { | ||
| const signaturePolicy = this.globalSignaturePolicy | ||
| switch (signaturePolicy) { | ||
| case SignaturePolicy.StrictSign: | ||
| // @ts-ignore seqno is optional in protobuf definition but it will exist | ||
|
There was a problem hiding this comment. Protobuf definition in libp2p specs has this as optional, however it is basically an outdated thing... |
||
| return utils.msgId(msg.from, msg.seqno) | ||
| case SignaturePolicy.StrictNoSign: | ||
| return utils.noSignMsgId(msg.data) | ||
|
|
@@ -508,25 +513,25 @@ class PubsubBaseProtocol extends EventEmitter { | |
| * @returns {RPC} | ||
| */ | ||
| _decodeRpc (bytes) { | ||
| return RPC.decode(bytes) | ||
| } | ||
|
|
||
| /** | ||
| * Encode RPC object into a Uint8Array. | ||
| * This can be override to use a custom router protobuf. | ||
| * | ||
| * @param {IRPC} rpc | ||
| * @returns {Uint8Array} | ||
| */ | ||
| _encodeRpc (rpc) { | ||
| return RPC.encode(rpc).finish() | ||
| } | ||
|
|
||
| /** | ||
| * Send an rpc object to a peer | ||
| * | ||
| * @param {string} id - peer id | ||
| * @param {IRPC} rpc | ||
| * @returns {void} | ||
| */ | ||
| _sendRpc (id, rpc) { | ||
|
|
@@ -592,12 +597,12 @@ class PubsubBaseProtocol extends EventEmitter { | |
| default: | ||
| throw errcode(new Error('Cannot validate message: unhandled signature policy: ' + signaturePolicy), codes.ERR_UNHANDLED_SIGNATURE_POLICY) | ||
| } | ||
|
|
||
| for (const topic of message.topicIDs) { | ||
| const validatorFn = this.topicValidators.get(topic) | ||
| if (validatorFn) { | ||
| await validatorFn(topic, message) | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -606,8 +611,8 @@ class PubsubBaseProtocol extends EventEmitter { | |
| * Should be used by the routers to create the message to send. | ||
| * | ||
| * @protected | ||
| * @param {InMessage} message | ||
| * @returns {Promise<InMessage>} | ||
| */ | ||
| _buildMessage (message) { | ||
| const signaturePolicy = this.globalSignaturePolicy | ||
|
|
@@ -617,7 +622,7 @@ class PubsubBaseProtocol extends EventEmitter { | |
| message.seqno = utils.randomSeqno() | ||
| return signMessage(this.peerId, utils.normalizeOutRpcMessage(message)) | ||
| case SignaturePolicy.StrictNoSign: | ||
| return Promise.resolve(message) | ||
| default: | ||
| throw errcode(new Error('Cannot build message: unhandled signature policy: ' + signaturePolicy), codes.ERR_UNHANDLED_SIGNATURE_POLICY) | ||
| } | ||
|
|
@@ -663,29 +668,30 @@ class PubsubBaseProtocol extends EventEmitter { | |
| this.log('publish', topic, message) | ||
|
|
||
| const from = this.peerId.toB58String() | ||
| const msgObject = { | ||
| receivedFrom: from, | ||
| data: message, | ||
| topicIDs: [topic] | ||
| } | ||
|
|
||
| // ensure that the message follows the signature policy | ||
| const outMsg = await this._buildMessage(msgObject) | ||
| // @ts-ignore different type as from is converted | ||
|
There was a problem hiding this comment. Because of the signatures, we have this "hack" where we also add the receivedFrom property and from is converted to a different type from the protobuf. Of couse ts does not like this, but I did not want to change this inner logic in this PR |
||
| const msg = utils.normalizeInRpcMessage(outMsg) | ||
|
|
||
| // Emit to self if I'm interested and emitSelf enabled | ||
| this.emitSelf && this._emitMessage(msg) | ||
|
|
||
| // send to all the other peers | ||
| await this._publish(msg) | ||
| } | ||
|
|
||
| /** | ||
| * Overriding the implementation of publish should handle the appropriate algorithms for the publish/subscriber implementation. | ||
| * For example, a Floodsub implementation might simply publish each message to each topic for every peer | ||
| * | ||
| * @abstract | ||
| * @param {InMessage|RPCMessage} message | ||
| * @returns {Promise<void>} | ||
| * | ||
| */ | ||
|
|
@@ -744,7 +750,6 @@ class PubsubBaseProtocol extends EventEmitter { | |
| } | ||
| } | ||
|
|
||
| PubsubBaseProtocol.utils = utils | ||
| PubsubBaseProtocol.SignaturePolicy = SignaturePolicy | ||
|
|
||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
probably should be "^5.0.0" now, Alex updated it and it was released yesterday, bl@5 has been out for a week or so too thanks to Alex as well
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
bl seems like a weird thing to use at the interface level, perhaps it's something we can pull out in future?
It's only used in one place and from the type def it looks like it's only necessary because the exported BufferList can't be treated as a Uint8Array based on the type heirachy.