fluencelabs/js-libp2p-interfaces
1
0
Fork 0
mirror of https://github.com/fluencelabs/js-libp2p-interfaces synced 2025-04-24 16:32:32 +00:00
Code Issues Projects Releases Wiki Activity

feat: add types

Browse Source
This commit is contained in:
Vasco Santos 2020-12-01 12:24:51 +01:00
parent 7fd26cf6b9
commit 6a95834570
25 changed files with 726 additions and 248 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
package.json
View File

@ -12,15 +12,15 @@
"scripts": {
"lint": "aegir lint",
"build": "aegir build",
"pregenerate:types": "rimraf './src/**/*.d.ts'",
"generate:types": "tsc --build",
"test": "aegir test",
"test:node": "aegir test --target node",
"test:browser": "aegir test --target browser",
"test:types": "aegir ts -p check",
"prepublishOnly": "npm run generate:types",
"release": "aegir release -t node -t browser",
"release-minor": "aegir release --type minor -t node -t browser",
"release-major": "aegir release --type major -t node -t browser"
"release-major": "aegir release --type major -t node -t browser",
"remove:types": "rimraf './src/**/*.d.ts'"
},
"repository": {
"type": "git",
@ -63,13 +63,13 @@
"protons": "^2.0.0",
"sinon": "^9.0.2",
"streaming-iterables": "^5.0.2",
"typescript": "^4.1.2",
"uint8arrays": "^1.1.0"
},
"devDependencies": {
"aegir": "^25.0.0",
"aegir": "^29.2.0",
"it-handshake": "^1.0.1",
"rimraf": "^3.0.2",
"typescript": "^4.0.5"
"rimraf": "^3.0.2"
},
"contributors": [
"Alan Shaw <alan.shaw@protocol.ai>",

224
src/connection/connection.d.ts vendored
View File

@ -1,4 +1,27 @@
export = Connection;
/**
* @typedef {import('../stream-muxer/types').MuxedStream} MuxedStream
*/
/**
* @typedef {Object} ConectionStat
* @property {string} direction - connection establishment direction ("inbound" or "outbound").
* @property {object} timeline - connection relevant events timestamp.
* @property {number} timeline.open - connection opening timestamp.
* @property {number} [timeline.upgraded] - connection upgraded timestamp.
* @property {number} [timeline.close] - connection upgraded timestamp.
* @property {string} [multiplexer] - connection multiplexing identifier.
* @property {string} [encryption] - connection encryption method identifier.
*
* @typedef {Object} ConnectionOptions
* @property {multiaddr} [localAddr] - local multiaddr of the connection if known.
* @property {multiaddr} remoteAddr - remote multiaddr of the connection.
* @property {PeerId} localPeer - local peer-id.
* @property {PeerId} remotePeer - remote peer-id.
* @property {(protocols: string|string[]) => Promise<{stream: MuxedStream, protocol: string}>} newStream - new stream muxer function.
* @property {() => Promise<void>} close - close raw connection function.
* @property {() => MuxedStream[]} getStreams - get streams from muxer function.
* @property {ConectionStat} stat - metadata of the connection.
*/
/**
* An implementation of the js-libp2p connection.
* Any libp2p transport should use an upgrader to return this connection.
@ -12,41 +35,13 @@ declare class Connection {
*/
static isConnection(other: any): other is Connection;
/**
* Creates an instance of Connection.
* @param {object} properties properties of the connection.
* @param {multiaddr} [properties.localAddr] local multiaddr of the connection if known.
* @param {multiaddr} [properties.remoteAddr] remote multiaddr of the connection.
* @param {PeerId} properties.localPeer local peer-id.
* @param {PeerId} properties.remotePeer remote peer-id.
* @param {function} properties.newStream new stream muxer function.
* @param {function} properties.close close raw connection function.
* @param {function(): Stream[]} properties.getStreams get streams from muxer function.
* @param {object} properties.stat metadata of the connection.
* @param {string} properties.stat.direction connection establishment direction ("inbound" or "outbound").
* @param {object} properties.stat.timeline connection relevant events timestamp.
* @param {string} properties.stat.timeline.open connection opening timestamp.
* @param {string} properties.stat.timeline.upgraded connection upgraded timestamp.
* @param {string} [properties.stat.multiplexer] connection multiplexing identifier.
* @param {string} [properties.stat.encryption] connection encryption method identifier.
* An implementation of the js-libp2p connection.
* Any libp2p transport should use an upgrader to return this connection.
*
* @class
* @param {ConnectionOptions} options
*/
constructor({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }: {
localAddr: multiaddr | undefined;
remoteAddr: multiaddr | undefined;
localPeer: PeerId;
remotePeer: PeerId;
newStream: Function;
close: Function;
getStreams: () => any[];
stat: {
direction: string;
timeline: {
open: string;
upgraded: string;
};
multiplexer: string | undefined;
encryption: string | undefined;
};
});
constructor({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }: ConnectionOptions);
/**
* Connection identifier.
*/
@ -54,107 +49,210 @@ declare class Connection {
/**
* Observed multiaddr of the local peer
*/
localAddr: multiaddr | undefined;
localAddr: import("multiaddr") | undefined;
/**
* Observed multiaddr of the remote peer
*/
remoteAddr: multiaddr | undefined;
remoteAddr: import("multiaddr");
/**
* Local peer id.
*/
localPeer: PeerId;
localPeer: import("peer-id");
/**
* Remote peer id.
*/
remotePeer: PeerId;
remotePeer: import("peer-id");
/**
* Connection metadata.
*/
_stat: {
status: "open";
/**
* - connection establishment direction ("inbound" or "outbound").
*/
direction: string;
/**
* - connection relevant events timestamp.
*/
timeline: {
open: string;
upgraded: string;
open: number;
upgraded: number | undefined;
close: number | undefined;
};
/**
* - connection multiplexing identifier.
*/
multiplexer?: string | undefined;
/**
* - connection encryption method identifier.
*/
encryption?: string | undefined;
};
/**
* Reference to the new stream function of the multiplexer
*/
_newStream: Function;
_newStream: (protocols: string | string[]) => Promise<{
stream: MuxedStream;
protocol: string;
}>;
/**
* Reference to the close function of the raw connection
*/
_close: Function;
_close: () => Promise<void>;
/**
* Reference to the getStreams function of the muxer
*/
_getStreams: () => any[];
_getStreams: () => MuxedStream[];
/**
* Connection streams registry
*/
registry: Map<any, any>;
/**
* User provided tags
*
* @type {string[]}
*/
tags: string[];
get [Symbol.toStringTag](): string;
/**
* Get connection metadata
*
* @this {Connection}
*/
get stat(): {
status: "open";
/**
* - connection establishment direction ("inbound" or "outbound").
*/
direction: string;
/**
* - connection relevant events timestamp.
*/
timeline: {
open: string;
upgraded: string;
open: number;
upgraded: number | undefined;
close: number | undefined;
};
/**
* - connection multiplexing identifier.
*/
multiplexer?: string | undefined;
/**
* - connection encryption method identifier.
*/
encryption?: string | undefined;
};
/**
* Get all the streams of the muxer.
*
* @this {Connection}
*/
get streams(): any[];
get streams(): import("../stream-muxer/types").MuxedStream[];
/**
* Create a new stream from this connection
* @param {string[]} protocols intended protocol for the stream
* @return {Promise<{stream: Stream, protocol: string}>} with muxed+multistream-selected stream and selected protocol
*
* @param {string|string[]} protocols - intended protocol for the stream
* @returns {Promise<{stream: MuxedStream, protocol: string}>} with muxed+multistream-selected stream and selected protocol
*/
newStream(protocols: string[]): Promise<{
stream: any;
newStream(protocols: string | string[]): Promise<{
stream: MuxedStream;
protocol: string;
}>;
/**
* Add a stream when it is opened to the registry.
* @param {*} muxedStream a muxed stream
* @param {object} properties the stream properties to be registered
* @param {string} properties.protocol the protocol used by the stream
* @param {object} properties.metadata metadata of the stream
* @return {void}
*
* @param {MuxedStream} muxedStream - a muxed stream
* @param {object} properties - the stream properties to be registered
* @param {string} properties.protocol - the protocol used by the stream
* @param {object} properties.metadata - metadata of the stream
* @returns {void}
*/
addStream(muxedStream: any, { protocol, metadata }: {
addStream(muxedStream: MuxedStream, { protocol, metadata }: {
protocol: string;
metadata: object;
}): void;
/**
* Remove stream registry after it is closed.
* @param {string} id identifier of the stream
*
* @param {string} id - identifier of the stream
*/
removeStream(id: string): void;
/**
* Close the connection.
* @return {Promise<void>}
*
* @returns {Promise<void>}
*/
close(): Promise<void>;
_closing: any;
get [connectionSymbol](): boolean;
_closing: void | undefined;
}
import multiaddr = require("multiaddr");
import PeerId = require("peer-id");
declare const connectionSymbol: unique symbol;
declare namespace Connection {
export { MuxedStream, ConectionStat, ConnectionOptions };
}
type MuxedStream = {
close: () => void;
abort: () => void;
reset: () => void;
sink: (source: Uint8Array) => Promise<Uint8Array>;
source: () => AsyncIterable<Uint8Array>;
timeline: import("../stream-muxer/types").MuxedTimeline;
id: string;
};
type ConnectionOptions = {
/**
* - local multiaddr of the connection if known.
*/
localAddr?: import("multiaddr") | undefined;
/**
* - remote multiaddr of the connection.
*/
remoteAddr: import("multiaddr");
/**
* - local peer-id.
*/
localPeer: import("peer-id");
/**
* - remote peer-id.
*/
remotePeer: import("peer-id");
/**
* - new stream muxer function.
*/
newStream: (protocols: string | string[]) => Promise<{
stream: MuxedStream;
protocol: string;
}>;
/**
* - close raw connection function.
*/
close: () => Promise<void>;
/**
* - get streams from muxer function.
*/
getStreams: () => MuxedStream[];
/**
* - metadata of the connection.
*/
stat: ConectionStat;
};
type ConectionStat = {
/**
* - connection establishment direction ("inbound" or "outbound").
*/
direction: string;
/**
* - connection relevant events timestamp.
*/
timeline: {
open: number;
upgraded: number | undefined;
close: number | undefined;
};
/**
* - connection multiplexing identifier.
*/
multiplexer?: string | undefined;
/**
* - connection encryption method identifier.
*/
encryption?: string | undefined;
};

159
src/connection/connection.js
View File

@ -8,51 +8,30 @@ const Status = require('./status')
const connectionSymbol = Symbol.for('@libp2p/interface-connection/connection')
function validateArgs (localAddr, localPeer, remotePeer, newStream, close, getStreams, stat) {
if (localAddr && !multiaddr.isMultiaddr(localAddr)) {
throw errCode(new Error('localAddr must be an instance of multiaddr'), 'ERR_INVALID_PARAMETERS')
}
/**
* @typedef {import('../stream-muxer/types').MuxedStream} MuxedStream
*/
if (!PeerId.isPeerId(localPeer)) {
throw errCode(new Error('localPeer must be an instance of peer-id'), 'ERR_INVALID_PARAMETERS')
}
if (!PeerId.isPeerId(remotePeer)) {
throw errCode(new Error('remotePeer must be an instance of peer-id'), 'ERR_INVALID_PARAMETERS')
}
if (typeof newStream !== 'function') {
throw errCode(new Error('new stream must be a function'), 'ERR_INVALID_PARAMETERS')
}
if (typeof close !== 'function') {
throw errCode(new Error('close must be a function'), 'ERR_INVALID_PARAMETERS')
}
if (typeof getStreams !== 'function') {
throw errCode(new Error('getStreams must be a function'), 'ERR_INVALID_PARAMETERS')
}
if (!stat) {
throw errCode(new Error('connection metadata object must be provided'), 'ERR_INVALID_PARAMETERS')
}
if (stat.direction !== 'inbound' && stat.direction !== 'outbound') {
throw errCode(new Error('direction must be "inbound" or "outbound"'), 'ERR_INVALID_PARAMETERS')
}
if (!stat.timeline) {
throw errCode(new Error('connection timeline object must be provided in the stat object'), 'ERR_INVALID_PARAMETERS')
}
if (!stat.timeline.open) {
throw errCode(new Error('connection open timestamp must be provided'), 'ERR_INVALID_PARAMETERS')
}
if (!stat.timeline.upgraded) {
throw errCode(new Error('connection upgraded timestamp must be provided'), 'ERR_INVALID_PARAMETERS')
}
}
/**
* @typedef {Object} ConectionStat
* @property {string} direction - connection establishment direction ("inbound" or "outbound").
* @property {object} timeline - connection relevant events timestamp.
* @property {number} timeline.open - connection opening timestamp.
* @property {number} [timeline.upgraded] - connection upgraded timestamp.
* @property {number} [timeline.close] - connection upgraded timestamp.
* @property {string} [multiplexer] - connection multiplexing identifier.
* @property {string} [encryption] - connection encryption method identifier.
*
* @typedef {Object} ConnectionOptions
* @property {multiaddr} [localAddr] - local multiaddr of the connection if known.
* @property {multiaddr} remoteAddr - remote multiaddr of the connection.
* @property {PeerId} localPeer - local peer-id.
* @property {PeerId} remotePeer - remote peer-id.
* @property {(protocols: string|string[]) => Promise<{stream: MuxedStream, protocol: string}>} newStream - new stream muxer function.
* @property {() => Promise<void>} close - close raw connection function.
* @property {() => MuxedStream[]} getStreams - get streams from muxer function.
* @property {ConectionStat} stat - metadata of the connection.
*/
/**
* An implementation of the js-libp2p connection.
@ -60,22 +39,11 @@ function validateArgs (localAddr, localPeer, remotePeer, newStream, close, getSt
*/
class Connection {
/**
* Creates an instance of Connection.
* @param {object} properties properties of the connection.
* @param {multiaddr} [properties.localAddr] local multiaddr of the connection if known.
* @param {multiaddr} [properties.remoteAddr] remote multiaddr of the connection.
* @param {PeerId} properties.localPeer local peer-id.
* @param {PeerId} properties.remotePeer remote peer-id.
* @param {function} properties.newStream new stream muxer function.
* @param {function} properties.close close raw connection function.
* @param {function(): Stream[]} properties.getStreams get streams from muxer function.
* @param {object} properties.stat metadata of the connection.
* @param {string} properties.stat.direction connection establishment direction ("inbound" or "outbound").
* @param {object} properties.stat.timeline connection relevant events timestamp.
* @param {string} properties.stat.timeline.open connection opening timestamp.
* @param {string} properties.stat.timeline.upgraded connection upgraded timestamp.
* @param {string} [properties.stat.multiplexer] connection multiplexing identifier.
* @param {string} [properties.stat.encryption] connection encryption method identifier.
* An implementation of the js-libp2p connection.
* Any libp2p transport should use an upgrader to return this connection.
*
* @class
* @param {ConnectionOptions} options
*/
constructor ({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }) {
validateArgs(localAddr, localPeer, remotePeer, newStream, close, getStreams, stat)
@ -135,6 +103,7 @@ class Connection {
/**
* User provided tags
*
* @type {string[]}
*/
this.tags = []
@ -160,6 +129,7 @@ class Connection {
/**
* Get connection metadata
*
* @this {Connection}
*/
get stat () {
@ -168,6 +138,7 @@ class Connection {
/**
* Get all the streams of the muxer.
*
* @this {Connection}
*/
get streams () {
@ -176,8 +147,9 @@ class Connection {
/**
* Create a new stream from this connection
* @param {string[]} protocols intended protocol for the stream
* @return {Promise<{stream: Stream, protocol: string}>} with muxed+multistream-selected stream and selected protocol
*
* @param {string|string[]} protocols - intended protocol for the stream
* @returns {Promise<{stream: MuxedStream, protocol: string}>} with muxed+multistream-selected stream and selected protocol
*/
async newStream (protocols) {
if (this.stat.status === Status.CLOSING) {
@ -202,11 +174,12 @@ class Connection {
/**
* Add a stream when it is opened to the registry.
* @param {*} muxedStream a muxed stream
* @param {object} properties the stream properties to be registered
* @param {string} properties.protocol the protocol used by the stream
* @param {object} properties.metadata metadata of the stream
* @return {void}
*
* @param {MuxedStream} muxedStream - a muxed stream
* @param {object} properties - the stream properties to be registered
* @param {string} properties.protocol - the protocol used by the stream
* @param {object} properties.metadata - metadata of the stream
* @returns {void}
*/
addStream (muxedStream, { protocol, metadata = {} }) {
// Add metadata for the stream
@ -218,7 +191,8 @@ class Connection {
/**
* Remove stream registry after it is closed.
* @param {string} id identifier of the stream
*
* @param {string} id - identifier of the stream
*/
removeStream (id) {
this.registry.delete(id)
@ -226,7 +200,8 @@ class Connection {
/**
* Close the connection.
* @return {Promise<void>}
*
* @returns {Promise<void>}
*/
async close () {
if (this.stat.status === Status.CLOSED) {
@ -248,3 +223,49 @@ class Connection {
}
module.exports = Connection
function validateArgs (localAddr, localPeer, remotePeer, newStream, close, getStreams, stat) {
if (localAddr && !multiaddr.isMultiaddr(localAddr)) {
throw errCode(new Error('localAddr must be an instance of multiaddr'), 'ERR_INVALID_PARAMETERS')
}
if (!PeerId.isPeerId(localPeer)) {
throw errCode(new Error('localPeer must be an instance of peer-id'), 'ERR_INVALID_PARAMETERS')
}
if (!PeerId.isPeerId(remotePeer)) {
throw errCode(new Error('remotePeer must be an instance of peer-id'), 'ERR_INVALID_PARAMETERS')
}
if (typeof newStream !== 'function') {
throw errCode(new Error('new stream must be a function'), 'ERR_INVALID_PARAMETERS')
}
if (typeof close !== 'function') {
throw errCode(new Error('close must be a function'), 'ERR_INVALID_PARAMETERS')
}
if (typeof getStreams !== 'function') {
throw errCode(new Error('getStreams must be a function'), 'ERR_INVALID_PARAMETERS')
}
if (!stat) {
throw errCode(new Error('connection metadata object must be provided'), 'ERR_INVALID_PARAMETERS')
}
if (stat.direction !== 'inbound' && stat.direction !== 'outbound') {
throw errCode(new Error('direction must be "inbound" or "outbound"'), 'ERR_INVALID_PARAMETERS')
}
if (!stat.timeline) {
throw errCode(new Error('connection timeline object must be provided in the stat object'), 'ERR_INVALID_PARAMETERS')
}
if (!stat.timeline.open) {
throw errCode(new Error('connection open timestamp must be provided'), 'ERR_INVALID_PARAMETERS')
}
if (!stat.timeline.upgraded) {
throw errCode(new Error('connection upgraded timestamp must be provided'), 'ERR_INVALID_PARAMETERS')
}
}

0
src/content-routing/types.ts Normal file
View File

40
src/crypto/types.ts Normal file
View File

@ -0,0 +1,40 @@
/**
* A libp2p crypto module must be compliant to this interface
* to ensure all exchanged data between two peers is encrypted.
*/
export interface CryptoInterface {
/**
* Encrypt outgoing data to the remote party.
*
* @param {PeerId} localPeer - PeerId of the receiving peer
* @param {MultiaddrConnection} connection - streaming iterable duplex that will be encrypted
* @param {PeerId} remotePeer - PeerId of the remote peer. Used to validate the integrity of the remote peer.
* @returns {Promise<SecureOutbound>}
*/
secureOutbound(localPeer: PeerId, connection: MultiaddrConnection, remotePeer: PeerId): Promise<SecureOutbound>;
/**
* Decrypt incoming data.
*
* @param {PeerId} localPeer - PeerId of the receiving peer.
* @param {MultiaddrConnection} connection - streaming iterable duplex that will be encryption.
* @param {PeerId} remotePeer - optional PeerId of the initiating peer, if known. This may only exist during transport upgrades.
* @returns {Promise<SecureOutbound>}
*/
secureInbound(localPeer: PeerId, connection: MultiaddrConnection, remotePeer?: PeerId): Promise<SecureOutbound>;
}
export declare class Crypto implements CryptoInterface {
protocol: string;
secureOutbound(localPeer: PeerId, connection: MultiaddrConnection, remotePeer: PeerId): Promise<SecureOutbound>;
secureInbound(localPeer: PeerId, connection: MultiaddrConnection, remotePeer?: PeerId): Promise<SecureOutbound>;
}
export type SecureOutbound = {
conn: MultiaddrConnection;
remoteEarlyData: Buffer;
remotePeer: PeerId;
}
type PeerId = import('peer-id');
type MultiaddrConnection = import('../transport/types').MultiaddrConnection

92
src/pubsub/index.d.ts vendored
View File

@ -13,28 +13,25 @@ export = PubsubBaseProtocol;
* @type import('peer-id')
*/
/**
* PubsubBaseProtocol handles the peers and connections logic for pubsub routers
* and specifies the API that pubsub routers should have.
*/
* PubsubBaseProtocol handles the peers and connections logic for pubsub routers
* and specifies the API that pubsub routers should have.
*/
declare class PubsubBaseProtocol {
/**
* @param {Object} props
* @param {String} props.debugName log namespace
* @param {Array<string>|string} props.multicodecs protocol identificers to connect
* @param {string} props.debugName - log namespace
* @param {Array<string>|string} props.multicodecs - protocol identificers to connect
* @param {Libp2p} props.libp2p
* @param {SignaturePolicy} [props.globalSignaturePolicy = SignaturePolicy.StrictSign] defines how signatures should be handled
* @param {boolean} [props.canRelayMessage = false] if can relay messages not subscribed
* @param {boolean} [props.emitSelf = false] if publish should emit to self, if subscribed
* @param {SignaturePolicy} [props.globalSignaturePolicy = SignaturePolicy.StrictSign] - defines how signatures should be handled
* @param {boolean} [props.canRelayMessage = false] - if can relay messages not subscribed
* @param {boolean} [props.emitSelf = false] - if publish should emit to self, if subscribed
* @abstract
*/
constructor({ debugName, multicodecs, libp2p, globalSignaturePolicy, canRelayMessage, emitSelf }: {
debugName: string;
multicodecs: Array<string> | string;
libp2p: any;
globalSignaturePolicy: {
StrictSign: "StrictSign";
StrictNoSign: string;
} | undefined;
globalSignaturePolicy: any;
canRelayMessage: boolean | undefined;
emitSelf: boolean | undefined;
});
@ -58,6 +55,7 @@ declare class PubsubBaseProtocol {
topics: Map<string, Set<string>>;
/**
* List of our subscriptions
*
* @type {Set<string>}
*/
subscriptions: Set<string>;
@ -75,16 +73,19 @@ declare class PubsubBaseProtocol {
globalSignaturePolicy: string;
/**
* If router can relay received messages, even if not subscribed
*
* @type {boolean}
*/
canRelayMessage: boolean;
/**
* if publish should emit to self, if subscribed
*
* @type {boolean}
*/
emitSelf: boolean;
/**
* Topic validator function
*
* @typedef {function(string, InMessage): Promise<void>} validator
*/
/**
@ -92,45 +93,52 @@ declare class PubsubBaseProtocol {
*
* Keyed by topic
* Topic validators are functions with the following input:
*
* @type {Map<string, validator>}
*/
topicValidators: Map<string, (arg0: string, arg1: InMessage) => Promise<void>>;
_registrarId: any;
/**
* On an inbound stream opened.
*
* @private
* @param {Object} props
* @param {string} props.protocol
* @param {DuplexIterableStream} props.stream
* @param {Connection} props.connection connection
* @param {Connection} props.connection - connection
*/
private _onIncomingStream;
/**
* Registrar notifies an established connection with pubsub protocol.
*
* @private
* @param {PeerId} peerId remote peer-id
* @param {Connection} conn connection to the peer
* @param {PeerId} peerId - remote peer-id
* @param {Connection} conn - connection to the peer
*/
private _onPeerConnected;
/**
* Registrar notifies a closing connection with pubsub protocol.
*
* @private
* @param {PeerId} peerId peerId
* @param {Error} err error for connection end
* @param {PeerId} peerId - peerId
* @param {Error} err - error for connection end
*/
private _onPeerDisconnected;
/**
* Register the pubsub protocol onto the libp2p node.
*
* @returns {void}
*/
start(): void;
/**
* Unregister the pubsub protocol and the streams with other peers will be closed.
*
* @returns {void}
*/
stop(): void;
/**
* Notifies the router that a peer has been connected
*
* @private
* @param {PeerId} peerId
* @param {string} protocol
@ -139,6 +147,7 @@ declare class PubsubBaseProtocol {
private _addPeer;
/**
* Notifies the router that a peer has been disconnected.
*
* @private
* @param {PeerId} peerId
* @returns {PeerStreams | undefined}
@ -146,47 +155,54 @@ declare class PubsubBaseProtocol {
private _removePeer;
/**
* Responsible for processing each RPC message received by other peers.
* @param {string} idB58Str peer id string in base58
* @param {DuplexIterableStream} stream inbound stream
* @param {PeerStreams} peerStreams PubSub peer
*
* @param {string} idB58Str - peer id string in base58
* @param {DuplexIterableStream} stream - inbound stream
* @param {PeerStreams} peerStreams - PubSub peer
* @returns {Promise<void>}
*/
_processMessages(idB58Str: string, stream: any, peerStreams: PeerStreams): Promise<void>;
_processMessages(idB58Str: string, stream: any, peerStreams: import("./peer-streams")): Promise<void>;
/**
* Handles an rpc request from a peer
* @param {String} idB58Str
*
* @param {string} idB58Str
* @param {PeerStreams} peerStreams
* @param {RPC} rpc
* @returns {boolean}
*/
_processRpc(idB58Str: string, peerStreams: PeerStreams, rpc: any): boolean;
_processRpc(idB58Str: string, peerStreams: import("./peer-streams"), rpc: any): boolean;
/**
* Handles a subscription change from a peer
*
* @param {string} id
* @param {RPC.SubOpt} subOpt
*/
_processRpcSubOpt(id: string, subOpt: any): void;
/**
* Handles an message from a peer
*
* @param {InMessage} msg
* @returns {Promise<void>}
*/
_processRpcMessage(msg: InMessage): Promise<void>;
/**
* Emit a message from a peer
*
* @param {InMessage} message
*/
_emitMessage(message: InMessage): void;
/**
* The default msgID implementation
* Child class can override this.
* @param {RPC.Message} msg the message object
*
* @param {RPC.Message} msg - the message object
* @returns {Uint8Array} message id as bytes
*/
getMsgId(msg: any): Uint8Array;
/**
* Whether to accept a message from a peer
* Override to create a graylist
*
* @override
* @param {string} id
* @returns {boolean}
@ -195,6 +211,7 @@ declare class PubsubBaseProtocol {
/**
* Decode Uint8Array into an RPC object.
* This can be override to use a custom router protobuf.
*
* @param {Uint8Array} bytes
* @returns {RPC}
*/
@ -202,28 +219,32 @@ declare class PubsubBaseProtocol {
/**
* Encode RPC object into a Uint8Array.
* This can be override to use a custom router protobuf.
*
* @param {RPC} rpc
* @returns {Uint8Array}
*/
_encodeRpc(rpc: any): Uint8Array;
/**
* Send an rpc object to a peer
* @param {string} id peer id
*
* @param {string} id - peer id
* @param {RPC} rpc
* @returns {void}
*/
_sendRpc(id: string, rpc: any): void;
/**
* Send subscroptions to a peer
* @param {string} id peer id
*
* @param {string} id - peer id
* @param {string[]} topics
* @param {boolean} subscribe set to false for unsubscriptions
* @param {boolean} subscribe - set to false for unsubscriptions
* @returns {void}
*/
_sendSubscriptions(id: string, topics: string[], subscribe: boolean): void;
/**
* Validates the given message. The signature will be checked for authenticity.
* Throws an error on invalid messages
*
* @param {InMessage} message
* @returns {Promise<void>}
*/
@ -231,6 +252,7 @@ declare class PubsubBaseProtocol {
/**
* Normalizes the message and signs it, if signing is enabled.
* Should be used by the routers to create the message to send.
*
* @private
* @param {Message} message
* @returns {Promise<Message>}
@ -238,12 +260,14 @@ declare class PubsubBaseProtocol {
private _buildMessage;
/**
* Get a list of the peer-ids that are subscribed to one topic.
*
* @param {string} topic
* @returns {Array<string>}
*/
getSubscribers(topic: string): Array<string>;
/**
* Publishes messages to all subscribed peers
*
* @override
* @param {string} topic
* @param {Buffer} message
@ -253,6 +277,7 @@ declare class PubsubBaseProtocol {
/**
* 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} message
* @returns {Promise<void>}
@ -261,6 +286,7 @@ declare class PubsubBaseProtocol {
_publish(message: InMessage): Promise<void>;
/**
* Subscribes to a given topic.
*
* @abstract
* @param {string} topic
* @returns {void}
@ -268,6 +294,7 @@ declare class PubsubBaseProtocol {
subscribe(topic: string): void;
/**
* Unsubscribe from the given topic.
*
* @override
* @param {string} topic
* @returns {void}
@ -275,8 +302,9 @@ declare class PubsubBaseProtocol {
unsubscribe(topic: string): void;
/**
* Get the list of topics which the peer is subscribed to.
*
* @override
* @returns {Array<String>}
* @returns {Array<string>}
*/
getTopics(): Array<string>;
}
@ -293,10 +321,12 @@ type InMessage = {
signature?: Uint8Array | undefined;
key?: Uint8Array | undefined;
};
import PeerStreams = require("./peer-streams");
/**
* @type {typeof import('./message')}
*/
declare const message: typeof import('./message');
import utils = require("./utils");
import { SignaturePolicy } from "./signature-policy";
declare const utils: typeof import("./utils");
declare const SignaturePolicy: {
StrictSign: "StrictSign";
StrictNoSign: string;
};

75
src/pubsub/index.js
View File

@ -37,18 +37,18 @@ const {
*/
/**
* PubsubBaseProtocol handles the peers and connections logic for pubsub routers
* and specifies the API that pubsub routers should have.
*/
* PubsubBaseProtocol handles the peers and connections logic for pubsub routers
* and specifies the API that pubsub routers should have.
*/
class PubsubBaseProtocol extends EventEmitter {
/**
* @param {Object} props
* @param {String} props.debugName log namespace
* @param {Array<string>|string} props.multicodecs protocol identificers to connect
* @param {string} props.debugName - log namespace
* @param {Array<string>|string} props.multicodecs - protocol identificers to connect
* @param {Libp2p} props.libp2p
* @param {SignaturePolicy} [props.globalSignaturePolicy = SignaturePolicy.StrictSign] defines how signatures should be handled
* @param {boolean} [props.canRelayMessage = false] if can relay messages not subscribed
* @param {boolean} [props.emitSelf = false] if publish should emit to self, if subscribed
* @param {SignaturePolicy} [props.globalSignaturePolicy = SignaturePolicy.StrictSign] - defines how signatures should be handled
* @param {boolean} [props.canRelayMessage = false] - if can relay messages not subscribed
* @param {boolean} [props.emitSelf = false] - if publish should emit to self, if subscribed
* @abstract
*/
constructor ({
@ -98,6 +98,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* List of our subscriptions
*
* @type {Set<string>}
*/
this.subscriptions = new Set()
@ -123,18 +124,21 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* If router can relay received messages, even if not subscribed
*
* @type {boolean}
*/
this.canRelayMessage = canRelayMessage
/**
* if publish should emit to self, if subscribed
*
* @type {boolean}
*/
this.emitSelf = emitSelf
/**
* Topic validator function
*
* @typedef {function(string, InMessage): Promise<void>} validator
*/
/**
@ -142,6 +146,7 @@ class PubsubBaseProtocol extends EventEmitter {
*
* Keyed by topic
* Topic validators are functions with the following input:
*
* @type {Map<string, validator>}
*/
this.topicValidators = new Map()
@ -156,6 +161,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Register the pubsub protocol onto the libp2p node.
*
* @returns {void}
*/
start () {
@ -185,6 +191,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Unregister the pubsub protocol and the streams with other peers will be closed.
*
* @returns {void}
*/
stop () {
@ -206,11 +213,12 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* On an inbound stream opened.
*
* @private
* @param {Object} props
* @param {string} props.protocol
* @param {DuplexIterableStream} props.stream
* @param {Connection} props.connection connection
* @param {Connection} props.connection - connection
*/
_onIncomingStream ({ protocol, stream, connection }) {
const peerId = connection.remotePeer
@ -223,9 +231,10 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Registrar notifies an established connection with pubsub protocol.
*
* @private
* @param {PeerId} peerId remote peer-id
* @param {Connection} conn connection to the peer
* @param {PeerId} peerId - remote peer-id
* @param {Connection} conn - connection to the peer
*/
async _onPeerConnected (peerId, conn) {
const idB58Str = peerId.toB58String()
@ -245,9 +254,10 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Registrar notifies a closing connection with pubsub protocol.
*
* @private
* @param {PeerId} peerId peerId
* @param {Error} err error for connection end
* @param {PeerId} peerId - peerId
* @param {Error} err - error for connection end
*/
_onPeerDisconnected (peerId, err) {
const idB58Str = peerId.toB58String()
@ -258,6 +268,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Notifies the router that a peer has been connected
*
* @private
* @param {PeerId} peerId
* @param {string} protocol
@ -288,6 +299,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Notifies the router that a peer has been disconnected.
*
* @private
* @param {PeerId} peerId
* @returns {PeerStreams | undefined}
@ -318,9 +330,10 @@ 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 {PeerStreams} peerStreams PubSub peer
*
* @param {string} idB58Str - peer id string in base58
* @param {DuplexIterableStream} stream - inbound stream
* @param {PeerStreams} peerStreams - PubSub peer
* @returns {Promise<void>}
*/
async _processMessages (idB58Str, stream, peerStreams) {
@ -343,7 +356,8 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Handles an rpc request from a peer
* @param {String} idB58Str
*
* @param {string} idB58Str
* @param {PeerStreams} peerStreams
* @param {RPC} rpc
* @returns {boolean}
@ -379,6 +393,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Handles a subscription change from a peer
*
* @param {string} id
* @param {RPC.SubOpt} subOpt
*/
@ -402,6 +417,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Handles an message from a peer
*
* @param {InMessage} msg
* @returns {Promise<void>}
*/
@ -426,6 +442,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Emit a message from a peer
*
* @param {InMessage} message
*/
_emitMessage (message) {
@ -439,7 +456,8 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* The default msgID implementation
* Child class can override this.
* @param {RPC.Message} msg the message object
*
* @param {RPC.Message} msg - the message object
* @returns {Uint8Array} message id as bytes
*/
getMsgId (msg) {
@ -457,6 +475,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Whether to accept a message from a peer
* Override to create a graylist
*
* @override
* @param {string} id
* @returns {boolean}
@ -468,6 +487,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Decode Uint8Array into an RPC object.
* This can be override to use a custom router protobuf.
*
* @param {Uint8Array} bytes
* @returns {RPC}
*/
@ -478,6 +498,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Encode RPC object into a Uint8Array.
* This can be override to use a custom router protobuf.
*
* @param {RPC} rpc
* @returns {Uint8Array}
*/
@ -487,7 +508,8 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Send an rpc object to a peer
* @param {string} id peer id
*
* @param {string} id - peer id
* @param {RPC} rpc
* @returns {void}
*/
@ -504,9 +526,10 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Send subscroptions to a peer
* @param {string} id peer id
*
* @param {string} id - peer id
* @param {string[]} topics
* @param {boolean} subscribe set to false for unsubscriptions
* @param {boolean} subscribe - set to false for unsubscriptions
* @returns {void}
*/
_sendSubscriptions (id, topics, subscribe) {
@ -518,6 +541,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Validates the given message. The signature will be checked for authenticity.
* Throws an error on invalid messages
*
* @param {InMessage} message
* @returns {Promise<void>}
*/
@ -564,6 +588,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Normalizes the message and signs it, if signing is enabled.
* Should be used by the routers to create the message to send.
*
* @private
* @param {Message} message
* @returns {Promise<Message>}
@ -586,6 +611,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Get a list of the peer-ids that are subscribed to one topic.
*
* @param {string} topic
* @returns {Array<string>}
*/
@ -607,6 +633,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Publishes messages to all subscribed peers
*
* @override
* @param {string} topic
* @param {Buffer} message
@ -640,6 +667,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* 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} message
* @returns {Promise<void>}
@ -651,6 +679,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Subscribes to a given topic.
*
* @abstract
* @param {string} topic
* @returns {void}
@ -668,6 +697,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Unsubscribe from the given topic.
*
* @override
* @param {string} topic
* @returns {void}
@ -685,8 +715,9 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Get the list of topics which the peer is subscribed to.
*
* @override
* @returns {Array<String>}
* @returns {Array<string>}
*/
getTopics () {
if (!this.started) {

9
src/pubsub/message/index.d.ts vendored
View File

@ -1,4 +1,5 @@
declare const rpcProto: any;
declare const topicDescriptorProto: any;
export const RPC: any;
export { rpcProto as rpc, topicDescriptorProto as td };
export var rpc: any;
export var td: any;
export var RPC: any;
export var Message: any;
export var SubOpts: any;

6
src/pubsub/message/sign.d.ts vendored
View File

@ -13,12 +13,12 @@ export function messagePublicKey(message: any): Promise<any>;
* @param {Message} message
* @returns {Promise<Message>}
*/
export function signMessage(peerId: PeerId, message: any): Promise<any>;
export function signMessage(peerId: import("peer-id"), message: any): Promise<any>;
export const SignPrefix: any;
/**
* Verifies the signature of the given message
*
* @param {InMessage} message
* @returns {Promise<Boolean>}
* @returns {Promise<boolean>}
*/
export function verifySignature(message: any): Promise<boolean>;
import PeerId = require("peer-id");

3
src/pubsub/message/sign.js
View File

@ -31,8 +31,9 @@ async function signMessage (peerId, message) {
/**
* Verifies the signature of the given message
*
* @param {InMessage} message
* @returns {Promise<Boolean>}
* @returns {Promise<boolean>}
*/
async function verifySignature (message) {
// Get message sans the signature

9
src/pubsub/peer-streams.d.ts vendored
View File

@ -16,7 +16,7 @@ export = PeerStreams;
*/
declare class PeerStreams {
/**
* @param {object} properties properties of the PeerStreams.
* @param {object} properties - properties of the PeerStreams.
* @param {PeerId} properties.id
* @param {string} properties.protocol
*/
@ -30,34 +30,40 @@ declare class PeerStreams {
id: import('peer-id');
/**
* Established protocol
*
* @type {string}
*/
protocol: string;
/**
* The raw outbound stream, as retrieved from conn.newStream
*
* @private
* @type {DuplexIterableStream}
*/
private _rawOutboundStream;
/**
* The raw inbound stream, as retrieved from the callback from libp2p.handle
*
* @private
* @type {DuplexIterableStream}
*/
private _rawInboundStream;
/**
* An AbortController for controlled shutdown of the inbound stream
*
* @private
* @type {typeof AbortController}
*/
private _inboundAbortController;
/**
* Write stream -- its preferable to use the write method
*
* @type {import('it-pushable').Pushable<Uint8Array>>}
*/
outboundStream: import('it-pushable').Pushable<Uint8Array>;
/**
* Read stream
*
* @type {DuplexIterableStream}
*/
inboundStream: DuplexIterableStream;
@ -97,6 +103,7 @@ declare class PeerStreams {
attachOutboundStream(stream: any): Promise<void>;
/**
* Closes the open connection to peer
*
* @returns {void}
*/
close(): void;

9
src/pubsub/peer-streams.js
View File

@ -30,7 +30,7 @@ log.error = debug('libp2p-pubsub:peer-streams:error')
*/
class PeerStreams extends EventEmitter {
/**
* @param {object} properties properties of the PeerStreams.
* @param {object} properties - properties of the PeerStreams.
* @param {PeerId} properties.id
* @param {string} properties.protocol
*/
@ -43,34 +43,40 @@ class PeerStreams extends EventEmitter {
this.id = id
/**
* Established protocol
*
* @type {string}
*/
this.protocol = protocol
/**
* The raw outbound stream, as retrieved from conn.newStream
*
* @private
* @type {DuplexIterableStream}
*/
this._rawOutboundStream = null
/**
* The raw inbound stream, as retrieved from the callback from libp2p.handle
*
* @private
* @type {DuplexIterableStream}
*/
this._rawInboundStream = null
/**
* An AbortController for controlled shutdown of the inbound stream
*
* @private
* @type {typeof AbortController}
*/
this._inboundAbortController = null
/**
* Write stream -- its preferable to use the write method
*
* @type {import('it-pushable').Pushable<Uint8Array>>}
*/
this.outboundStream = null
/**
* Read stream
*
* @type {DuplexIterableStream}
*/
this.inboundStream = null
@ -179,6 +185,7 @@ class PeerStreams extends EventEmitter {
/**
* Closes the open connection to peer
*
* @returns {void}
*/
close () {

4
src/pubsub/utils.js
View File

@ -91,7 +91,7 @@ exports.ensureArray = (maybeArray) => {
* @template {Object} T
* @param {T} message
* @param {string} [peerId]
* @return {T & {from?: string, peerId?: string }}
* @returns {T & {from?: string, peerId?: string }}
*/
exports.normalizeInRpcMessage = (message, peerId) => {
const m = Object.assign({}, message)
@ -108,7 +108,7 @@ exports.normalizeInRpcMessage = (message, peerId) => {
* @template {Object} T
*
* @param {T} message
* @return {T & {from?: Uint8Array, data?: Uint8Array}}
* @returns {T & {from?: Uint8Array, data?: Uint8Array}}
*/
exports.normalizeOutRpcMessage = (message) => {
const m = Object.assign({}, message)

13
src/record/index.d.ts vendored
View File

@ -4,20 +4,23 @@ export = Record;
*/
declare class Record {
/**
* @constructor
* @param {String} domain signature domain
* @param {Uint8Array} codec identifier of the type of record
* @class
* @param {string} domain - signature domain
* @param {Uint8Array} codec - identifier of the type of record
*/
constructor(domain: string, codec: Uint8Array);
domain: string;
codec: Uint8Array;
/**
* Marshal a record to be used in an envelope.
* @returns {Uint8Array}
*/
marshal(): void;
marshal(): Uint8Array;
/**
* Verifies if the other provided Record is identical to this one.
*
* @param {Record} other
* @returns {boolean}
*/
equals(other: Record): void;
equals(other: Record): boolean;
}

12
src/record/index.js
View File

@ -7,25 +7,31 @@ const errcode = require('err-code')
*/
class Record {
/**
* @constructor
* @param {String} domain signature domain
* @param {Uint8Array} codec identifier of the type of record
* @class
* @param {string} domain - signature domain
* @param {Uint8Array} codec - identifier of the type of record
*/
constructor (domain, codec) {
this.domain = domain
this.codec = codec
}
// eslint-disable-next-line
/**
* Marshal a record to be used in an envelope.
*
* @returns {Uint8Array}
*/
marshal () {
throw errcode(new Error('marshal must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
}
// eslint-disable-next-line
/**
* Verifies if the other provided Record is identical to this one.
*
* @param {Record} other
* @returns {boolean}
*/
equals (other) {
throw errcode(new Error('equals must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')

5
src/stream-muxer/tests/base-test.js
View File

@ -20,8 +20,9 @@ async function closeAndWait (stream) {
/**
* A tick is considered valid if it happened between now
* and `ms` milliseconds ago
* @param {number} date Time in ticks
* @param {number} ms max milliseconds that should have expired
*
* @param {number} date - Time in ticks
* @param {number} ms - max milliseconds that should have expired
* @returns {boolean}
*/
function isValidTick (date, ms = 5000) {

38
src/stream-muxer/types.d.ts vendored Normal file
View File

@ -0,0 +1,38 @@
/**
* A libp2p stream muxer
*/
export interface StreamMuxerInterface {
readonly streams: Array<MuxedStream>;
/**
* Initiate a new stream with the given name. If no name is
* provided, the id of th stream will be used.
*
* @param {string} [name] - If name is not a string it will be cast to one
* @returns {Stream}
*/
newStream(name?: string): MuxedStream;
onStream(stream: MuxedStream): void;
onStreamEnd(stream: MuxedStream): void;
}
export declare class Muxer implements StreamMuxerInterface {
multicodec: string;
readonly streams: Array<MuxedStream>;
newStream(name?: string): MuxedStream;
onStream(stream: MuxedStream): void;
onStreamEnd(stream: MuxedStream): void;
}
export declare type MuxedTimeline = {
open: number;
close?: number;
};
export declare type MuxedStream = {
close: () => void;
abort: () => void;
reset: () => void;
sink: Sink;
source: () => AsyncIterable<Uint8Array>;
timeline: MuxedTimeline;
id: string;
};
declare type Sink = (source: Uint8Array) => Promise<Uint8Array>;
export {};

53
src/stream-muxer/types.ts Normal file
View File

@ -0,0 +1,53 @@
/**
* A libp2p stream muxer
*/
export interface StreamMuxerInterface {
readonly streams: Array<MuxedStream>;
/**
* Initiate a new stream with the given name. If no name is
* provided, the id of th stream will be used.
*
* @param {string} [name] - If name is not a string it will be cast to one
* @returns {Stream}
*/
newStream (name?: string): MuxedStream;
/**
* A function called when receiving a new stream from the remote.
*
* @param {MuxedStream} stream
*/
onStream (stream: MuxedStream): void;
/**
* A function called when a stream ends.
*
* @param {MuxedStream} stream
*/
onStreamEnd (stream: MuxedStream): void;
}
export declare class Muxer implements StreamMuxerInterface {
multicodec: string;
readonly streams: Array<MuxedStream>;
newStream (name?: string): MuxedStream;
onStream(stream: MuxedStream): void;
onStreamEnd(stream: MuxedStream): void;
}
export type MuxedTimeline = {
open: number;
close?: number;
}
export type MuxedStream = {
close: () => void;
abort: () => void;
reset: () => void;
sink: Sink;
source: () => AsyncIterable<Uint8Array>;
timeline: MuxedTimeline;
id: string;
}
type Sink = (source: Uint8Array) => Promise<Uint8Array>;

32
src/topology/index.d.ts vendored
View File

@ -1,4 +1,17 @@
export = Topology;
/**
* @typedef {import('peer-id')} PeerId
*/
/**
* @typedef {Object} Options
* @property {number} [min=0] - minimum needed connections.
* @property {number} [max=Infinity] - maximum needed connections.
* @property {Handlers} [handlers]
*
* @typedef {Object} Handlers
* @property {(peerId: PeerId, conn: import('../connection')) => void} [onConnect] - protocol "onConnect" handler
* @property {(peerId: PeerId) => void} [onDisconnect] - protocol "onDisconnect" handler
*/
declare class Topology {
/**
* Checks if the given value is a Topology instance.
@ -13,31 +26,30 @@ declare class Topology {
constructor({ min, max, handlers }: Options);
min: number;
max: number;
_onConnect: Function;
_onDisconnect: Function;
_onConnect: (peerId: PeerId, conn: import('../connection')) => void;
_onDisconnect: (peerId: PeerId) => void;
/**
* Set of peers that support the protocol.
*
* @type {Set<string>}
*/
peers: Set<string>;
get [Symbol.toStringTag](): string;
set registrar(arg: any);
_registrar: any;
/**
* @typedef PeerId
* @type {import('peer-id')}
*/
/**
* Notify about peer disconnected event.
*
* @param {PeerId} peerId
* @returns {void}
*/
disconnect(peerId: import("peer-id")): void;
disconnect(peerId: PeerId): void;
get [topologySymbol](): boolean;
}
declare namespace Topology {
export { Options, Handlers };
export { PeerId, Options, Handlers };
}
type PeerId = import("peer-id");
declare const topologySymbol: unique symbol;
type Options = {
/**
@ -54,9 +66,9 @@ type Handlers = {
/**
* - protocol "onConnect" handler
*/
onConnect?: Function | undefined;
onConnect?: ((peerId: PeerId, conn: import('../connection')) => void) | undefined;
/**
* - protocol "onDisconnect" handler
*/
onDisconnect?: Function | undefined;
onDisconnect?: ((peerId: PeerId) => void) | undefined;
};

35
src/topology/index.js
View File

@ -4,6 +4,21 @@
const noop = () => {}
const topologySymbol = Symbol.for('@libp2p/js-interfaces/topology')
/**
* @typedef {import('peer-id')} PeerId
*/
/**
* @typedef {Object} Options
* @property {number} [min=0] - minimum needed connections.
* @property {number} [max=Infinity] - maximum needed connections.
* @property {Handlers} [handlers]
*
* @typedef {Object} Handlers
* @property {(peerId: PeerId, conn: import('../connection')) => void} [onConnect] - protocol "onConnect" handler
* @property {(peerId: PeerId) => void} [onDisconnect] - protocol "onDisconnect" handler
*/
class Topology {
/**
* @param {Options} options
@ -22,6 +37,7 @@ class Topology {
/**
* Set of peers that support the protocol.
*
* @type {Set<string>}
*/
this.peers = new Set()
@ -45,17 +61,13 @@ class Topology {
return Boolean(other && other[topologySymbol])
}
set registrar (registrar) {
set registrar (registrar) { // eslint-disable-line
this._registrar = registrar
}
/**
* @typedef PeerId
* @type {import('peer-id')}
*/
/**
* Notify about peer disconnected event.
*
* @param {PeerId} peerId
* @returns {void}
*/
@ -64,15 +76,4 @@ class Topology {
}
}
/**
* @typedef {Object} Options
* @property {number} [min=0] - minimum needed connections.
* @property {number} [max=Infinity] - maximum needed connections.
* @property {Handlers} [handlers]
*
* @typedef {Object} Handlers
* @property {Function} [onConnect] - protocol "onConnect" handler
* @property {Function} [onDisconnect] - protocol "onDisconnect" handler
*/
module.exports = Topology

13
src/topology/multicodec-topology.d.ts vendored
View File

@ -1,5 +1,6 @@
export = MulticodecTopology;
declare class MulticodecTopology extends Topology {
declare const MulticodecTopology_base: typeof import(".");
declare class MulticodecTopology extends MulticodecTopology_base {
/**
* Checks if the given value is a `MulticodecTopology` instance.
*
@ -14,6 +15,7 @@ declare class MulticodecTopology extends Topology {
multicodecs: string[];
/**
* Check if a new peer support the multicodecs for this topology.
*
* @param {Object} props
* @param {PeerId} props.peerId
* @param {Array<string>} props.protocols
@ -24,12 +26,14 @@ declare class MulticodecTopology extends Topology {
}): void;
/**
* Verify if a new connected peer has a topology multicodec and call _onConnect.
*
* @param {Connection} connection
* @returns {void}
*/
_onPeerConnect(connection: Connection): void;
/**
* Update topology.
*
* @param {Array<{id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string>}>} peerDataIterable
* @returns {void}
*/
@ -43,7 +47,6 @@ declare class MulticodecTopology extends Topology {
declare namespace MulticodecTopology {
export { PeerId, Multiaddr, Connection, TopologyOptions, MulticodecOptions, Handlers };
}
import Topology = require(".");
type PeerId = import("peer-id");
type Connection = typeof import("../connection");
type Multiaddr = import("multiaddr");
@ -57,7 +60,7 @@ type TopologyOptions = {
* - maximum needed connections.
*/
max?: number | undefined;
handlers?: Topology.Handlers | undefined;
handlers?: import(".").Handlers | undefined;
};
type MulticodecOptions = {
/**
@ -70,9 +73,9 @@ type Handlers = {
/**
* - protocol "onConnect" handler
*/
onConnect?: Function | undefined;
onConnect?: ((peerId: import("peer-id"), conn: typeof import("../connection")) => void) | undefined;
/**
* - protocol "onDisconnect" handler
*/
onDisconnect?: Function | undefined;
onDisconnect?: ((peerId: import("peer-id")) => void) | undefined;
};

5
src/topology/multicodec-topology.js
View File

@ -57,7 +57,7 @@ class MulticodecTopology extends Topology {
return Boolean(other && other[multicodecTopologySymbol])
}
set registrar (registrar) {
set registrar (registrar) { // eslint-disable-line
this._registrar = registrar
this._registrar.peerStore.on('change:protocols', this._onProtocolChange)
this._registrar.connectionManager.on('peer:connect', this._onPeerConnect)
@ -68,6 +68,7 @@ class MulticodecTopology extends Topology {
/**
* Update topology.
*
* @param {Array<{id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string>}>} peerDataIterable
* @returns {void}
*/
@ -88,6 +89,7 @@ class MulticodecTopology extends Topology {
/**
* Check if a new peer support the multicodecs for this topology.
*
* @param {Object} props
* @param {PeerId} props.peerId
* @param {Array<string>} props.protocols
@ -113,6 +115,7 @@ class MulticodecTopology extends Topology {
/**
* Verify if a new connected peer has a topology multicodec and call _onConnect.
*
* @param {Connection} connection
* @returns {void}
*/

5
src/transport/tests/utils/index.js
View File

@ -4,8 +4,9 @@ module.exports = {
/**
* A tick is considered valid if it happened between now
* and `ms` milliseconds ago
* @param {number} date Time in ticks
* @param {number} ms max milliseconds that should have expired
*
* @param {number} date - Time in ticks
* @param {number} ms - max milliseconds that should have expired
* @returns {boolean}
*/
isValidTick: function isValidTick (date, ms = 5000) {

120
src/transport/types.ts Normal file
View File

@ -0,0 +1,120 @@
import events from 'events'
/**
* A libp2p transport is understood as something that offers a dial and listen interface to establish connections.
*/
export interface Interface {
/**
* Dial a given multiaddr.
*
* @param {Multiaddr} ma
* @param {any} [options]
* @returns {Promise<Connection>}
*/
dial(ma: Multiaddr, options?: any): Promise<Connection>;
/**
* Create transport listeners.
*
* @param {any} options
* @param {(Connection) => void} handler
*/
createListener(options: any, handler: (Connection) => void): Listener;
/**
* Takes a list of `Multiaddr`s and returns only valid addresses for the transport
*
* @param {Multiaddr[]} multiaddrs
* @returns {Multiaddr[]}
*/
filter(multiaddrs: Multiaddr[]): Multiaddr[];
}
export interface Listener extends events.EventEmitter {
/**
* Start a listener
*
* @param {Multiaddr} multiaddr
* @returns {Promise<void>}
*/
listen(multiaddr: Multiaddr): Promise<void>;
/**
* Get listen addresses
*
* @returns {Multiaddr[]}
*/
getAddrs(): Multiaddr[];
/**
* Close listener
*
* @returns {Promise<void>}
*/
close(): Promise<void>;
}
export interface Upgrader {
/**
* Upgrades an outbound connection on `transport.dial`.
*
* @param {MultiaddrConnection} maConn
* @returns {Promise<Connection>}
*/
upgradeOutbound(maConn: MultiaddrConnection): Promise<Connection>;
/**
* Upgrades an inbound connection on transport listener.
*
* @param {MultiaddrConnection} maConn
* @returns {Promise<Connection>}
*/
upgradeInbound(maConn: MultiaddrConnection): Promise<Connection>;
}
export declare class Transport implements Interface {
constructor({ upgrader, ...others }: {
upgrader: Upgrader;
others: any;
});
/**
* Dial a given multiaddr.
*
* @param {Multiaddr} ma
* @param {any} [options]
* @returns {Promise<Connection>}
*/
dial(ma: Multiaddr, options?: any): Promise<Connection>;
/**
* Create transport listeners.
*
* @param {any} options
* @param {(Connection) => void} handler
*/
createListener(options: any, handler: (Connection) => void): Listener;
/**
* Takes a list of `Multiaddr`s and returns only valid addresses for the transport
*
* @param {Multiaddr[]} multiaddrs
* @returns {Multiaddr[]}
*/
filter(multiaddrs: Multiaddr[]): Multiaddr[];
}
export type MultiaddrConnectionTimeline = {
open: number;
upgraded?: number;
close?: number;
}
export type MultiaddrConnection = {
sink: Sink;
source: () => AsyncIterable<Uint8Array>;
close: (err?: Error) => Promise<void>;
conn: any;
remoteAddr: Multiaddr;
localAddr?: Multiaddr;
timeline: MultiaddrConnectionTimeline;
}
type Sink = (source: Uint8Array) => Promise<Uint8Array>;
type Connection = import('../connection/connection')
type Multiaddr = import('multiaddr');

1
test/connection/compliance.spec.js
View File

@ -13,6 +13,7 @@ describe('compliance tests', () => {
/**
* Test setup. `properties` allows the compliance test to override
* certain values for testing.
*
* @param {*} properties
*/
async setup (properties) {