fluencelabs/js-libp2p-interfaces
1
0
Fork 0
mirror of https://github.com/fluencelabs/js-libp2p-interfaces synced 2025-03-14 21:41:08 +00:00
Code Issues Projects Releases Wiki Activity

feat: add types (#74)

Browse Source
This commit is contained in:
Vasco Santos 2020-12-10 14:03:17 +01:00 committed by GitHub
parent 04e93d3f08
commit e2419ea308
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
50 changed files with 770 additions and 1204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
.github/workflows/main.yml vendored Normal file
View File

@ -0,0 +1,53 @@
name: ci
on:
push:
branches:
- master
pull_request:
branches:
- master
jobs:
check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: yarn lint
- uses: gozala/typescript-error-reporter-action@v1.0.4
- run: yarn build
- run: yarn aegir dep-check
- uses: ipfs/aegir/actions/bundle-size@master
name: size
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
test-node:
needs: check
runs-on: ${{ matrix.os }}
strategy:
matrix:
os: [windows-latest, ubuntu-latest, macos-latest]
node: [12, 14]
fail-fast: true
steps:
- uses: actions/checkout@v2
- uses: actions/setup-node@v1
with:
node-version: ${{ matrix.node }}
- run: yarn
- run: npx nyc --reporter=lcov aegir test -t node -- --bail
- uses: codecov/codecov-action@v1
test-chrome:
needs: check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: npx aegir test -t browser -t webworker --bail
test-firefox:
needs: check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: npx aegir test -t browser -t webworker --bail -- --browsers FirefoxHeadless

40
.travis.yml
View File

@ -1,40 +0,0 @@
language: node_js
cache: npm
stages:
- check
- test
- cov
node_js:
- '10'
- '12'
os:
- linux
- osx
- windows
script: npx nyc -s npm run test:node -- --bail
after_success: npx nyc report --reporter=text-lcov > coverage.lcov && npx codecov
jobs:
include:
- stage: check
script:
- npx aegir dep-check
- npm run lint
- stage: test
name: chrome
addons:
chrome: stable
script: npx aegir test -t browser -t webworker
- stage: test
name: firefox
addons:
firefox: latest
script: npx aegir test -t browser -t webworker -- --browsers FirefoxHeadless
notifications:
email: false

21
package.json
View File

@ -9,11 +9,21 @@
"types",
"dist"
],
"types": "dist/src/index.d.ts",
"typesVersions": {
"*": {
"src/*": [
"dist/src/*",
"dist/src/*/index"
]
}
},
"eslintConfig": {
"extends": "ipfs"
},
"scripts": {
"lint": "aegir lint",
"build": "aegir build",
"pregenerate:types": "rimraf './src/**/*.d.ts'",
"generate:types": "tsc",
"test": "aegir test",
"test:node": "aegir test --target node",
"test:browser": "aegir test --target browser",
@ -37,11 +47,11 @@
},
"homepage": "https://github.com/libp2p/js-interfaces#readme",
"dependencies": {
"@types/bl": "^2.1.0",
"abort-controller": "^3.0.0",
"abortable-iterator": "^3.0.0",
"chai": "^4.2.0",
"chai-checkmark": "^1.0.1",
"class-is": "^1.1.0",
"debug": "^4.1.1",
"delay": "^4.3.0",
"detect-node": "^2.0.4",
@ -67,10 +77,9 @@
"uint8arrays": "^1.1.0"
},
"devDependencies": {
"aegir": "^25.0.0",
"aegir": "^29.2.0",
"it-handshake": "^1.0.1",
"rimraf": "^3.0.2",
"typescript": "3.7.5"
"rimraf": "^3.0.2"
},
"contributors": [
"Alan Shaw <alan.shaw@protocol.ai>",

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

@ -1,149 +0,0 @@
declare const _exports: typeof Connection;
export = _exports;
/**
* An implementation of the js-libp2p connection.
* Any libp2p transport should use an upgrader to return this connection.
*/
declare 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.
*/
constructor({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }: {
localAddr?: import("multiaddr");
remoteAddr?: import("multiaddr");
localPeer: import("peer-id");
remotePeer: import("peer-id");
newStream: Function;
close: Function;
getStreams: () => any[];
stat: {
direction: string;
timeline: {
open: string;
upgraded: string;
};
multiplexer?: string;
encryption?: string;
};
});
/**
* Connection identifier.
*/
id: any;
/**
* Observed multiaddr of the local peer
*/
localAddr: import("multiaddr");
/**
* Observed multiaddr of the remote peer
*/
remoteAddr: import("multiaddr");
/**
* Local peer id.
*/
localPeer: import("peer-id");
/**
* Remote peer id.
*/
remotePeer: import("peer-id");
/**
* Connection metadata.
*/
_stat: {
status: string;
direction: string;
timeline: {
open: string;
upgraded: string;
};
multiplexer?: string;
encryption?: string;
};
/**
* Reference to the new stream function of the multiplexer
*/
_newStream: Function;
/**
* Reference to the close function of the raw connection
*/
_close: Function;
/**
* Reference to the getStreams function of the muxer
*/
_getStreams: () => any[];
/**
* Connection streams registry
*/
registry: Map<any, any>;
/**
* User provided tags
* @type {string[]}
*/
tags: string[];
/**
* Get connection metadata
* @this {Connection}
*/
get stat(): {
status: string;
direction: string;
timeline: {
open: string;
upgraded: string;
};
multiplexer?: string;
encryption?: string;
};
/**
* Get all the streams of the muxer.
* @this {Connection}
*/
get streams(): any[];
/**
* 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
*/
newStream(protocols: string[]): Promise<{
stream: any;
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}
*/
addStream(muxedStream: any, { protocol, metadata }: {
protocol: string;
metadata: any;
}): void;
/**
* Remove stream registry after it is closed.
* @param {string} id identifier of the stream
*/
removeStream(id: string): void;
/**
* Close the connection.
* @return {Promise<void>}
*/
close(): Promise<void>;
_closing: any;
}

409
src/connection/connection.js
View File

@ -2,9 +2,233 @@
const PeerId = require('peer-id')
const multiaddr = require('multiaddr')
const withIs = require('class-is')
const errCode = require('err-code')
const Status = require('./status')
const { OPEN, CLOSING, CLOSED } = require('./status')
const connectionSymbol = Symbol.for('@libp2p/interface-connection/connection')
/**
* @typedef {import('../stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('./status').Status} Status
*/
/**
* @typedef {Object} Timeline
* @property {number} open - connection opening timestamp.
* @property {number} [upgraded] - connection upgraded timestamp.
* @property {number} [close]
*
* @typedef {Object} ConectionStat
* @property {string} direction - connection establishment direction ("inbound" or "outbound").
* @property {Timeline} timeline - connection relevant events 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.
*
* @typedef {Object} StreamData
* @property {string} protocol - the protocol used by the stream
* @property {Object} [metadata] - metadata of the stream
*/
/**
* An implementation of the js-libp2p connection.
* Any libp2p transport should use an upgrader to return this connection.
*/
class Connection {
/**
* 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)
/**
* Connection identifier.
*/
this.id = (parseInt(String(Math.random() * 1e9))).toString(36) + Date.now()
/**
* Observed multiaddr of the local peer
*/
this.localAddr = localAddr
/**
* Observed multiaddr of the remote peer
*/
this.remoteAddr = remoteAddr
/**
* Local peer id.
*/
this.localPeer = localPeer
/**
* Remote peer id.
*/
this.remotePeer = remotePeer
/**
* Connection metadata.
*
* @type {ConectionStat & {status: Status}}
*/
this._stat = {
...stat,
status: OPEN
}
/**
* Reference to the new stream function of the multiplexer
*/
this._newStream = newStream
/**
* Reference to the close function of the raw connection
*/
this._close = close
/**
* Reference to the getStreams function of the muxer
*/
this._getStreams = getStreams
/**
* Connection streams registry
*/
this.registry = new Map()
/**
* User provided tags
*
* @type {string[]}
*/
this.tags = []
}
get [Symbol.toStringTag] () {
return 'Connection'
}
get [connectionSymbol] () {
return true
}
/**
* Checks if the given value is a `Connection` instance.
*
* @param {any} other
* @returns {other is Connection}
*/
static isConnection (other) {
return Boolean(other && other[connectionSymbol])
}
/**
* Get connection metadata
*
* @this {Connection}
*/
get stat () {
return this._stat
}
/**
* Get all the streams of the muxer.
*
* @this {Connection}
*/
get streams () {
return this._getStreams()
}
/**
* Create a new stream from this connection
*
* @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 === CLOSING) {
throw errCode(new Error('the connection is being closed'), 'ERR_CONNECTION_BEING_CLOSED')
}
if (this.stat.status === CLOSED) {
throw errCode(new Error('the connection is closed'), 'ERR_CONNECTION_CLOSED')
}
if (!Array.isArray(protocols)) protocols = [protocols]
const { stream, protocol } = await this._newStream(protocols)
this.addStream(stream, { protocol })
return {
stream,
protocol
}
}
/**
* Add a stream when it is opened to the registry.
*
* @param {MuxedStream} muxedStream - a muxed stream
* @param {StreamData} data - the stream data to be registered
* @returns {void}
*/
addStream (muxedStream, { protocol, metadata = {} }) {
// Add metadata for the stream
this.registry.set(muxedStream.id, {
protocol,
...metadata
})
}
/**
* Remove stream registry after it is closed.
*
* @param {string} id - identifier of the stream
*/
removeStream (id) {
this.registry.delete(id)
}
/**
* Close the connection.
*
* @returns {Promise<void>}
*/
async close () {
if (this.stat.status === CLOSED) {
return
}
if (this._closing) {
return this._closing
}
this.stat.status = CLOSING
// Close raw connection
this._closing = await this._close()
this._stat.timeline.close = Date.now()
this.stat.status = CLOSED
}
}
module.exports = Connection
function validateArgs (localAddr, localPeer, remotePeer, newStream, close, getStreams, stat) {
if (localAddr && !multiaddr.isMultiaddr(localAddr)) {
@ -51,184 +275,3 @@ function validateArgs (localAddr, localPeer, remotePeer, newStream, close, getSt
throw errCode(new Error('connection upgraded timestamp must be provided'), 'ERR_INVALID_PARAMETERS')
}
}
/**
* An implementation of the js-libp2p connection.
* Any libp2p transport should use an upgrader to return this connection.
*/
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.
*/
constructor ({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }) {
validateArgs(localAddr, localPeer, remotePeer, newStream, close, getStreams, stat)
/**
* Connection identifier.
*/
this.id = (parseInt(Math.random() * 1e9)).toString(36) + Date.now()
/**
* Observed multiaddr of the local peer
*/
this.localAddr = localAddr
/**
* Observed multiaddr of the remote peer
*/
this.remoteAddr = remoteAddr
/**
* Local peer id.
*/
this.localPeer = localPeer
/**
* Remote peer id.
*/
this.remotePeer = remotePeer
/**
* Connection metadata.
*/
this._stat = {
...stat,
status: Status.OPEN
}
/**
* Reference to the new stream function of the multiplexer
*/
this._newStream = newStream
/**
* Reference to the close function of the raw connection
*/
this._close = close
/**
* Reference to the getStreams function of the muxer
*/
this._getStreams = getStreams
/**
* Connection streams registry
*/
this.registry = new Map()
/**
* User provided tags
* @type {string[]}
*/
this.tags = []
}
/**
* Get connection metadata
* @this {Connection}
*/
get stat () {
return this._stat
}
/**
* Get all the streams of the muxer.
* @this {Connection}
*/
get streams () {
return this._getStreams()
}
/**
* 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
*/
async newStream (protocols) {
if (this.stat.status === Status.CLOSING) {
throw errCode(new Error('the connection is being closed'), 'ERR_CONNECTION_BEING_CLOSED')
}
if (this.stat.status === Status.CLOSED) {
throw errCode(new Error('the connection is closed'), 'ERR_CONNECTION_CLOSED')
}
if (!Array.isArray(protocols)) protocols = [protocols]
const { stream, protocol } = await this._newStream(protocols)
this.addStream(stream, { protocol })
return {
stream,
protocol
}
}
/**
* 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}
*/
addStream (muxedStream, { protocol, metadata = {} }) {
// Add metadata for the stream
this.registry.set(muxedStream.id, {
protocol,
...metadata
})
}
/**
* Remove stream registry after it is closed.
* @param {string} id identifier of the stream
*/
removeStream (id) {
this.registry.delete(id)
}
/**
* Close the connection.
* @return {Promise<void>}
*/
async close () {
if (this.stat.status === Status.CLOSED) {
return
}
if (this._closing) {
return this._closing
}
this.stat.status = Status.CLOSING
// Close raw connection
this._closing = await this._close()
this._stat.timeline.close = Date.now()
this.stat.status = Status.CLOSED
}
}
/**
* @module
* @type {typeof Connection}
*/
module.exports = withIs(Connection, { className: 'Connection', symbolName: '@libp2p/interface-connection/connection' })

1
src/connection/index.d.ts vendored
View File

@ -1 +0,0 @@
export var Connection: typeof import('./connection');

4
src/connection/index.js
View File

@ -1,7 +1,3 @@
'use strict'
/**
* @module connection/index
* @type {typeof import('./connection')}
*/
exports.Connection = require('./connection')

3
src/connection/status.d.ts vendored
View File

@ -1,3 +0,0 @@
export declare const OPEN: string;
export declare const CLOSING: string;
export declare const CLOSED: string;

13
src/connection/status.js
View File

@ -1,7 +1,12 @@
'use strict'
module.exports = {
OPEN: 'open',
CLOSING: 'closing',
CLOSED: 'closed'
const STATUS = {
OPEN: /** @type {'open'} */('open'),
CLOSING: /** @type {'closing'} */('closing'),
CLOSED: /** @type {'closed'} */('closed')
}
module.exports = STATUS
/**
* @typedef {STATUS[keyof STATUS]} Status
*/

3
src/connection/tests/connection.js
View File

@ -74,6 +74,7 @@ module.exports = (test) => {
let timelineProxy
const proxyHandler = {
set () {
// @ts-ignore - TS fails to infer here
return Reflect.set(...arguments)
}
}
@ -138,7 +139,9 @@ module.exports = (test) => {
expect(connection.stat.timeline.close).to.not.exist()
await connection.close()
// @ts-ignore - fails to infer callCount
expect(proxyHandler.set.callCount).to.equal(1)
// @ts-ignore - fails to infer getCall
const [obj, key, value] = proxyHandler.set.getCall(0).args
expect(obj).to.eql(connection.stat.timeline)
expect(key).to.equal('close')

0
src/index.d.ts → src/content-routing/types.ts
View File

15
src/crypto/errors.d.ts vendored
View File

@ -1,15 +0,0 @@
export class UnexpectedPeerError extends Error {
static get code(): string;
constructor(message?: string);
code: string;
}
export class InvalidCryptoExchangeError extends Error {
static get code(): string;
constructor(message?: string);
code: string;
}
export class InvalidCryptoTransmissionError extends Error {
static get code(): string;
constructor(message?: string);
code: string;
}

2
src/crypto/tests/index.js
View File

@ -6,7 +6,7 @@ const expect = chai.expect
chai.use(require('dirty-chai'))
const duplexPair = require('it-pair/duplex')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const PeerId = require('peer-id')
const { collect } = require('streaming-iterables')
const uint8arrayFromString = require('uint8arrays/from-string')

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

@ -0,0 +1,24 @@
import PeerId from 'peer-id'
import { MultiaddrConnection } from '../transport/types'
/**
* A libp2p crypto module must be compliant to this interface
* to ensure all exchanged data between two peers is encrypted.
*/
export interface Crypto {
protocol: string;
/**
* Encrypt outgoing data to the remote party.
*/
secureOutbound(localPeer: PeerId, connection: MultiaddrConnection, remotePeer: PeerId): Promise<SecureOutbound>;
/**
* Decrypt incoming data.
*/
secureInbound(localPeer: PeerId, connection: MultiaddrConnection, remotePeer?: PeerId): Promise<SecureOutbound>;
}
export type SecureOutbound = {
conn: MultiaddrConnection;
remoteEarlyData: Buffer;
remotePeer: PeerId;
}

11
src/pubsub/errors.d.ts vendored
View File

@ -1,11 +0,0 @@
export namespace codes {
export const ERR_INVALID_SIGNATURE_POLICY: string;
export const ERR_UNHANDLED_SIGNATURE_POLICY: string;
export const ERR_MISSING_SIGNATURE: string;
export const ERR_MISSING_SEQNO: string;
export const ERR_INVALID_SIGNATURE: string;
export const ERR_UNEXPECTED_FROM: string;
export const ERR_UNEXPECTED_SIGNATURE: string;
export const ERR_UNEXPECTED_KEY: string;
export const ERR_UNEXPECTED_SEQNO: string;
}

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

@ -1,309 +0,0 @@
export = PubsubBaseProtocol;
/**
* @typedef {Object} InMessage
* @property {string} [from]
* @property {string} receivedFrom
* @property {string[]} topicIDs
* @property {Uint8Array} [seqno]
* @property {Uint8Array} data
* @property {Uint8Array} [signature]
* @property {Uint8Array} [key]
*
* @typedef PeerId
* @type import('peer-id')
*/
/**
* 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 {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
* @abstract
*/
constructor({ debugName, multicodecs, libp2p, globalSignaturePolicy, canRelayMessage, emitSelf }: {
debugName: string;
multicodecs: string | string[];
libp2p: any;
globalSignaturePolicy?: any;
canRelayMessage?: boolean;
emitSelf?: boolean;
});
log: any;
/**
* @type {Array<string>}
*/
multicodecs: Array<string>;
_libp2p: any;
registrar: any;
/**
* @type {PeerId}
*/
peerId: PeerId;
started: boolean;
/**
* Map of topics to which peers are subscribed to
*
* @type {Map<string, Set<string>>}
*/
topics: Map<string, Set<string>>;
/**
* List of our subscriptions
* @type {Set<string>}
*/
subscriptions: Set<string>;
/**
* Map of peer streams
*
* @type {Map<string, import('./peer-streams')>}
*/
peers: Map<string, import('./peer-streams')>;
/**
* The signature policy to follow by default
*
* @type {string}
*/
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
*/
/**
* Topic validator map
*
* Keyed by topic
* Topic validators are functions with the following input:
* @type {Map<string, validator>}
*/
topicValidators: Map<string, validator>;
_registrarId: any;
/**
* On an inbound stream opened.
* @private
* @param {Object} props
* @param {string} props.protocol
* @param {DuplexIterableStream} props.stream
* @param {Connection} props.connection connection
*/
_onIncomingStream({ protocol, stream, connection }: {
protocol: string;
stream: any;
connection: any;
}): void;
/**
* Registrar notifies an established connection with pubsub protocol.
* @private
* @param {PeerId} peerId remote peer-id
* @param {Connection} conn connection to the peer
*/
_onPeerConnected(peerId: import("peer-id"), conn: any): Promise<void>;
/**
* Registrar notifies a closing connection with pubsub protocol.
* @private
* @param {PeerId} peerId peerId
* @param {Error} err error for connection end
*/
_onPeerDisconnected(peerId: import("peer-id"), err: Error): void;
/**
* 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
* @returns {PeerStreams}
*/
_addPeer(peerId: import("peer-id"), protocol: string): import("./peer-streams");
/**
* Notifies the router that a peer has been disconnected.
* @private
* @param {PeerId} peerId
* @returns {PeerStreams | undefined}
*/
_removePeer(peerId: import("peer-id")): import("./peer-streams");
/**
* 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
* @returns {Promise<void>}
*/
_processMessages(idB58Str: string, stream: any, peerStreams: import("./peer-streams")): Promise<void>;
/**
* Handles an rpc request from a peer
* @param {String} idB58Str
* @param {PeerStreams} peerStreams
* @param {RPC} rpc
* @returns {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
* @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}
*/
_acceptFrom(id: string): boolean;
/**
* Decode Uint8Array into an RPC object.
* This can be override to use a custom router protobuf.
* @param {Uint8Array} bytes
* @returns {RPC}
*/
_decodeRpc(bytes: Uint8Array): any;
/**
* 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 {RPC} rpc
* @returns {void}
*/
_sendRpc(id: string, rpc: any): void;
/**
* Send subscroptions to a peer
* @param {string} id peer id
* @param {string[]} topics
* @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>}
*/
validate(message: InMessage): Promise<void>;
/**
* 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>}
*/
_buildMessage(message: any): Promise<any>;
/**
* Get a list of the peer-ids that are subscribed to one topic.
* @param {string} topic
* @returns {Array<string>}
*/
getSubscribers(topic: string): string[];
/**
* Publishes messages to all subscribed peers
* @override
* @param {string} topic
* @param {Buffer} message
* @returns {Promise<void>}
*/
publish(topic: string, message: Buffer): Promise<void>;
/**
* 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>}
*
*/
_publish(message: InMessage): Promise<void>;
/**
* Subscribes to a given topic.
* @abstract
* @param {string} topic
* @returns {void}
*/
subscribe(topic: string): void;
/**
* Unsubscribe from the given topic.
* @override
* @param {string} topic
* @returns {void}
*/
unsubscribe(topic: string): void;
/**
* Get the list of topics which the peer is subscribed to.
* @override
* @returns {Array<String>}
*/
getTopics(): string[];
}
declare namespace PubsubBaseProtocol {
export { message, utils, SignaturePolicy, InMessage, PeerId };
}
type PeerId = import("peer-id");
/**
* Topic validator function
*/
type validator = (arg0: string, arg1: InMessage) => Promise<void>;
type InMessage = {
from?: string;
receivedFrom: string;
topicIDs: string[];
seqno?: Uint8Array;
data: Uint8Array;
signature?: Uint8Array;
key?: Uint8Array;
};
/**
* @type {typeof import('./message')}
*/
declare const message: typeof import('./message');
declare const utils: typeof import("./utils");
declare const SignaturePolicy: {
StrictSign: string;
StrictNoSign: string;
};

125
src/pubsub/index.js
View File

@ -1,10 +1,10 @@
'use strict'
const debug = require('debug')
const EventEmitter = require('events')
const { EventEmitter } = require('events')
const errcode = require('err-code')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const MulticodecTopology = require('../topology/multicodec-topology')
const { codes } = require('./errors')
@ -21,6 +21,18 @@ const {
verifySignature
} = require('./message/sign')
/**
* @typedef {any} Libp2p
* @typedef {import('peer-id')} PeerId
* @typedef {import('bl')} BufferList
* @typedef {import('../stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('../connection/connection')} Connection
* @typedef {import('./message').RPC} RPC
* @typedef {import('./message').SubOpts} RPCSubOpts
* @typedef {import('./message').Message} RPCMessage
* @typedef {import('./signature-policy').SignaturePolicyType} SignaturePolicyType
*/
/**
* @typedef {Object} InMessage
* @property {string} [from]
@ -30,24 +42,21 @@ const {
* @property {Uint8Array} data
* @property {Uint8Array} [signature]
* @property {Uint8Array} [key]
*
* @typedef PeerId
* @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.
*/
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 {SignaturePolicyType} [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 ({
@ -97,6 +106,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* List of our subscriptions
*
* @type {Set<string>}
*/
this.subscriptions = new Set()
@ -122,18 +132,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
*/
/**
@ -141,6 +154,7 @@ class PubsubBaseProtocol extends EventEmitter {
*
* Keyed by topic
* Topic validators are functions with the following input:
*
* @type {Map<string, validator>}
*/
this.topicValidators = new Map()
@ -155,6 +169,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Register the pubsub protocol onto the libp2p node.
*
* @returns {void}
*/
start () {
@ -184,6 +199,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Unregister the pubsub protocol and the streams with other peers will be closed.
*
* @returns {void}
*/
stop () {
@ -205,26 +221,28 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* On an inbound stream opened.
* @private
*
* @protected
* @param {Object} props
* @param {string} props.protocol
* @param {DuplexIterableStream} props.stream
* @param {Connection} props.connection connection
* @param {MuxedStream} props.stream
* @param {Connection} props.connection - connection
*/
_onIncomingStream ({ protocol, stream, connection }) {
const peerId = connection.remotePeer
const idB58Str = peerId.toB58String()
const peer = this._addPeer(peerId, protocol)
peer.attachInboundStream(stream)
const inboundStream = peer.attachInboundStream(stream)
this._processMessages(idB58Str, peer.inboundStream, peer)
this._processMessages(idB58Str, inboundStream, peer)
}
/**
* Registrar notifies an established connection with pubsub protocol.
* @private
* @param {PeerId} peerId remote peer-id
* @param {Connection} conn connection to the peer
*
* @protected
* @param {PeerId} peerId - remote peer-id
* @param {Connection} conn - connection to the peer
*/
async _onPeerConnected (peerId, conn) {
const idB58Str = peerId.toB58String()
@ -244,9 +262,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
*
* @protected
* @param {PeerId} peerId - peerId
* @param {Error} [err] - error for connection end
*/
_onPeerDisconnected (peerId, err) {
const idB58Str = peerId.toB58String()
@ -257,7 +276,8 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Notifies the router that a peer has been connected
* @private
*
* @protected
* @param {PeerId} peerId
* @param {string} protocol
* @returns {PeerStreams}
@ -287,7 +307,8 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Notifies the router that a peer has been disconnected.
* @private
*
* @protected
* @param {PeerId} peerId
* @returns {PeerStreams | undefined}
*/
@ -317,9 +338,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 {AsyncIterable<Uint8Array|BufferList>} stream - inbound stream
* @param {PeerStreams} peerStreams - PubSub peer
* @returns {Promise<void>}
*/
async _processMessages (idB58Str, stream, peerStreams) {
@ -342,7 +364,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}
@ -378,8 +401,9 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Handles a subscription change from a peer
*
* @param {string} id
* @param {RPC.SubOpt} subOpt
* @param {RPCSubOpts} subOpt
*/
_processRpcSubOpt (id, subOpt) {
const t = subOpt.topicID
@ -401,6 +425,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Handles an message from a peer
*
* @param {InMessage} msg
* @returns {Promise<void>}
*/
@ -425,6 +450,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Emit a message from a peer
*
* @param {InMessage} message
*/
_emitMessage (message) {
@ -438,7 +464,8 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* The default msgID implementation
* Child class can override this.
* @param {RPC.Message} msg the message object
*
* @param {RPCMessage} msg - the message object
* @returns {Uint8Array} message id as bytes
*/
getMsgId (msg) {
@ -456,6 +483,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Whether to accept a message from a peer
* Override to create a graylist
*
* @override
* @param {string} id
* @returns {boolean}
@ -467,6 +495,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}
*/
@ -477,6 +506,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}
*/
@ -486,7 +516,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}
*/
@ -503,9 +534,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) {
@ -517,6 +549,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>}
*/
@ -563,9 +596,10 @@ 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>}
*
* @protected
* @param {RPCMessage} message
* @returns {Promise<RPCMessage>}
*/
_buildMessage (message) {
const signaturePolicy = this.globalSignaturePolicy
@ -585,6 +619,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Get a list of the peer-ids that are subscribed to one topic.
*
* @param {string} topic
* @returns {Array<string>}
*/
@ -606,6 +641,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Publishes messages to all subscribed peers
*
* @override
* @param {string} topic
* @param {Buffer} message
@ -639,6 +675,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>}
@ -650,6 +687,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Subscribes to a given topic.
*
* @abstract
* @param {string} topic
* @returns {void}
@ -667,6 +705,7 @@ class PubsubBaseProtocol extends EventEmitter {
/**
* Unsubscribe from the given topic.
*
* @override
* @param {string} topic
* @returns {void}
@ -684,8 +723,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) {
@ -696,7 +736,8 @@ class PubsubBaseProtocol extends EventEmitter {
}
}
PubsubBaseProtocol.message = message
PubsubBaseProtocol.utils = utils
PubsubBaseProtocol.SignaturePolicy = SignaturePolicy
module.exports = PubsubBaseProtocol
module.exports.message = message
module.exports.utils = utils
module.exports.SignaturePolicy = SignaturePolicy

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

@ -1,5 +0,0 @@
export var rpc: any;
export var td: any;
export var RPC: any;
export var Message: any;
export var SubOpts: any;

2
src/pubsub/message/rpc.proto.d.ts vendored
View File

@ -1,2 +0,0 @@
declare const _exports: string;
export = _exports;

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

@ -1,23 +0,0 @@
/**
* Returns the PublicKey associated with the given message.
* If no, valid PublicKey can be retrieved an error will be returned.
*
* @param {InMessage} message
* @returns {Promise<PublicKey>}
*/
export function messagePublicKey(message: any): Promise<any>;
/**
* Signs the provided message with the given `peerId`
*
* @param {PeerId} peerId
* @param {Message} message
* @returns {Promise<Message>}
*/
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>}
*/
export function verifySignature(message: any): Promise<boolean>;

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

@ -31,18 +31,24 @@ 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) {
if (!message.signature) {
throw new Error('Message must contain a signature to be verified')
}
// Get message sans the signature
const baseMessage = { ...message }
delete baseMessage.signature
delete baseMessage.key
baseMessage.from = PeerId.createFromCID(baseMessage.from).toBytes()
const bytes = uint8ArrayConcat([
SignPrefix,
Message.encode(baseMessage)
Message.encode({
...message,
from: message.from && PeerId.createFromCID(message.from).toBytes(),
signature: undefined,
key: undefined
})
])
// Get the public key
@ -61,13 +67,17 @@ async function verifySignature (message) {
*/
async function messagePublicKey (message) {
// should be available in the from property of the message (peer id)
if (!message.from) {
throw new Error('Could not get the public key from the originator id')
}
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) {
@ -77,6 +87,11 @@ async function messagePublicKey (message) {
}
}
/**
* @typedef {import('..').InMessage} InMessage
* @typedef {import('libp2p-crypto').PublicKey} PublicKey
*/
module.exports = {
messagePublicKey,
signMessage,

2
src/pubsub/message/topic-descriptor.proto.d.ts vendored
View File

@ -1,2 +0,0 @@
declare const _exports: string;
export = _exports;

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

@ -1,113 +0,0 @@
export = PeerStreams;
/**
* @callback Sink
* @param {Uint8Array} source
* @returns {Promise<Uint8Array>}
*
* @typedef {object} DuplexIterableStream
* @property {Sink} sink
* @property {() AsyncIterator<Uint8Array>} source
*
* @typedef PeerId
* @type import('peer-id')
*/
/**
* Thin wrapper around a peer's inbound / outbound pubsub streams
*/
declare class PeerStreams {
/**
* @param {object} properties properties of the PeerStreams.
* @param {PeerId} properties.id
* @param {string} properties.protocol
*/
constructor({ id, protocol }: {
id: import("peer-id");
protocol: string;
});
/**
* @type {import('peer-id')}
*/
id: import('peer-id');
/**
* Established protocol
* @type {string}
*/
protocol: string;
/**
* The raw outbound stream, as retrieved from conn.newStream
* @private
* @type {DuplexIterableStream}
*/
_rawOutboundStream: DuplexIterableStream;
/**
* The raw inbound stream, as retrieved from the callback from libp2p.handle
* @private
* @type {DuplexIterableStream}
*/
_rawInboundStream: DuplexIterableStream;
/**
* An AbortController for controlled shutdown of the inbound stream
* @private
* @type {typeof AbortController}
*/
_inboundAbortController: typeof AbortController;
/**
* 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;
/**
* Do we have a connection to read from?
*
* @type {boolean}
*/
get isReadable(): boolean;
/**
* Do we have a connection to write on?
*
* @type {boolean}
*/
get isWritable(): boolean;
/**
* Send a message to this peer.
* Throws if there is no `stream` to write to available.
*
* @param {Uint8Array} data
* @returns {void}
*/
write(data: Uint8Array): void;
/**
* Attach a raw inbound stream and setup a read stream
*
* @param {DuplexIterableStream} stream
* @returns {void}
*/
attachInboundStream(stream: DuplexIterableStream): void;
/**
* Attach a raw outbound stream and setup a write stream
*
* @param {Stream} stream
* @returns {Promise<void>}
*/
attachOutboundStream(stream: any): Promise<void>;
/**
* Closes the open connection to peer
* @returns {void}
*/
close(): void;
}
declare namespace PeerStreams {
export { Sink, DuplexIterableStream, PeerId };
}
type DuplexIterableStream = {
sink: Sink;
source: () => AsyncIterator<Uint8Array, any, undefined>;
};
declare const AbortController: typeof import("abort-controller");
type Sink = (source: Uint8Array) => Promise<Uint8Array>;
type PeerId = import("peer-id");

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

@ -1,28 +1,24 @@
'use strict'
const EventEmitter = require('events')
const { EventEmitter } = require('events')
const lp = require('it-length-prefixed')
/** @type {typeof import('it-pushable').default} */
// @ts-ignore
const pushable = require('it-pushable')
const pipe = require('it-pipe')
const abortable = require('abortable-iterator')
const AbortController = require('abort-controller')
const { pipe } = require('it-pipe')
const { source: abortable } = require('abortable-iterator')
const AbortController = require('abort-controller').default
const debug = require('debug')
const log = debug('libp2p-pubsub:peer-streams')
log.error = debug('libp2p-pubsub:peer-streams:error')
/**
* @callback Sink
* @param {Uint8Array} source
* @returns {Promise<Uint8Array>}
*
* @typedef {object} DuplexIterableStream
* @property {Sink} sink
* @property {() AsyncIterator<Uint8Array>} source
*
* @typedef PeerId
* @type import('peer-id')
* @typedef {import('../stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('peer-id')} PeerId
* @typedef {import('it-pushable').Pushable<Uint8Array>} PushableStream
*/
/**
@ -30,7 +26,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,35 +39,41 @@ 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}
* @type {null|MuxedStream}
*/
this._rawOutboundStream = null
/**
* The raw inbound stream, as retrieved from the callback from libp2p.handle
*
* @private
* @type {DuplexIterableStream}
* @type {null|MuxedStream}
*/
this._rawInboundStream = null
/**
* An AbortController for controlled shutdown of the inbound stream
*
* @private
* @type {typeof AbortController}
* @type {AbortController}
*/
this._inboundAbortController = null
this._inboundAbortController = new AbortController()
/**
* Write stream -- its preferable to use the write method
* @type {import('it-pushable').Pushable<Uint8Array>>}
*
* @type {null|PushableStream}
*/
this.outboundStream = null
/**
* Read stream
* @type {DuplexIterableStream}
*
* @type {null| AsyncIterable<Uint8Array>}
*/
this.inboundStream = null
}
@ -102,7 +104,7 @@ class PeerStreams extends EventEmitter {
* @returns {void}
*/
write (data) {
if (!this.isWritable) {
if (!this.outboundStream) {
const id = this.id.toB58String()
throw new Error('No writable connection to ' + id)
}
@ -113,15 +115,14 @@ class PeerStreams extends EventEmitter {
/**
* Attach a raw inbound stream and setup a read stream
*
* @param {DuplexIterableStream} stream
* @returns {void}
* @param {MuxedStream} stream
* @returns {AsyncIterable<Uint8Array>}
*/
attachInboundStream (stream) {
// Create and attach a new inbound stream
// 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
this.inboundStream = abortable(
pipe(
@ -133,31 +134,31 @@ class PeerStreams extends EventEmitter {
)
this.emit('stream:inbound')
return this.inboundStream
}
/**
* Attach a raw outbound stream and setup a write stream
*
* @param {Stream} stream
* @param {MuxedStream} stream
* @returns {Promise<void>}
*/
async attachOutboundStream (stream) {
// If an outbound stream already exists,
// gently close it
// 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
await this.outboundStream.end(false)
await this.outboundStream.end()
}
this._rawOutboundStream = stream
this.outboundStream = pushable({
onEnd: (shouldEmit) => {
// close writable side of the stream
this._rawOutboundStream.reset && this._rawOutboundStream.reset()
this._rawOutboundStream && this._rawOutboundStream.reset && this._rawOutboundStream.reset()
this._rawOutboundStream = null
this.outboundStream = null
if (shouldEmit !== false) {
if (shouldEmit) {
this.emit('close')
}
}
@ -179,6 +180,7 @@ class PeerStreams extends EventEmitter {
/**
* Closes the open connection to peer
*
* @returns {void}
*/
close () {

4
src/pubsub/signature-policy.d.ts vendored
View File

@ -1,4 +0,0 @@
export namespace SignaturePolicy {
export const StrictSign: string;
export const StrictNoSign: string;
}

11
src/pubsub/signature-policy.js
View File

@ -4,7 +4,7 @@
* Enum for Signature Policy
* Details how message signatures are produced/consumed
*/
exports.SignaturePolicy = {
const SignaturePolicy = {
/**
* On the producing side:
* * Build messages with the signature, key (from may be enough for certain inlineable public key types), from and seqno fields.
@ -13,7 +13,7 @@ exports.SignaturePolicy = {
* * Enforce the fields to be present, reject otherwise.
* * Propagate only if the fields are valid and signature can be verified, reject otherwise.
*/
StrictSign: 'StrictSign',
StrictSign: /** @type {'StrictSign'} */ ('StrictSign'),
/**
* On the producing side:
* * Build messages without the signature, key, from and seqno fields.
@ -24,5 +24,10 @@ exports.SignaturePolicy = {
* * Propagate only if the fields are absent, reject otherwise.
* * A message_id function will not be able to use the above fields, and should instead rely on the data field. A commonplace strategy is to calculate a hash.
*/
StrictNoSign: 'StrictNoSign'
StrictNoSign: /** @type {'StrictNoSign'} */ ('StrictNoSign')
}
exports.SignaturePolicy = SignaturePolicy
/**
* @typedef {SignaturePolicy[keyof SignaturePolicy]} SignaturePolicyType
*/

2
src/pubsub/tests/emit-self.js
View File

@ -62,7 +62,7 @@ module.exports = (common) => {
pubsub.publish(topic, data)
// Wait 1 second to guarantee that self is not noticed
return new Promise((resolve) => setTimeout(() => resolve(), 1000))
return new Promise((resolve) => setTimeout(resolve, 1000))
})
})
})

11
src/pubsub/tests/messages.js
View File

@ -50,7 +50,10 @@ module.exports = (common) => {
sinon.spy(pubsub, '_publish')
sinon.spy(pubsub, 'validate')
const peerStream = new PeerStreams({ id: await PeerId.create() })
const peerStream = new PeerStreams({
id: await PeerId.create(),
protocol: 'test'
})
const rpc = {
subscriptions: [],
msgs: [{
@ -82,7 +85,11 @@ module.exports = (common) => {
sinon.spy(pubsub, '_publish')
sinon.spy(pubsub, 'validate')
const peerStream = new PeerStreams({ id: await PeerId.create() })
const peerStream = new PeerStreams({
id: await PeerId.create(),
protocol: 'test'
})
const rpc = {
subscriptions: [],
msgs: [{

12
src/pubsub/tests/multiple-nodes.js
View File

@ -113,9 +113,9 @@ module.exports = (common) => {
// await subscription change
await Promise.all([
new Promise(resolve => psA.once('pubsub:subscription-change', () => resolve())),
new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve())),
new Promise(resolve => psC.once('pubsub:subscription-change', () => resolve()))
new Promise(resolve => psA.once('pubsub:subscription-change', () => resolve(null))),
new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve(null))),
new Promise(resolve => psC.once('pubsub:subscription-change', () => resolve(null)))
])
// await a cycle
@ -166,9 +166,9 @@ module.exports = (common) => {
// await subscription change
await Promise.all([
new Promise(resolve => psA.once('pubsub:subscription-change', () => resolve())),
new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve())),
new Promise(resolve => psC.once('pubsub:subscription-change', () => resolve()))
new Promise(resolve => psA.once('pubsub:subscription-change', () => resolve(null))),
new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve(null))),
new Promise(resolve => psC.once('pubsub:subscription-change', () => resolve(null)))
])
psA.on(topic, incMsg)

7
src/pubsub/utils.d.ts vendored
View File

@ -1,7 +0,0 @@
export function randomSeqno(): Uint8Array;
export function msgId(from: string, seqno: Uint8Array): Uint8Array;
export function noSignMsgId(data: Uint8Array): Uint8Array;
export function anyMatch(a: any[] | Set<any>, b: any[] | Set<any>): boolean;
export function ensureArray(maybeArray: any): any[];
export function normalizeInRpcMessage(message: any, peerId: string): any;
export function normalizeOutRpcMessage(message: any): any;

19
src/pubsub/utils.js
View File

@ -71,8 +71,9 @@ exports.anyMatch = (a, b) => {
/**
* Make everything an array.
*
* @param {any} maybeArray
* @returns {Array}
* @template T
* @param {T|T[]} maybeArray
* @returns {T[]}
* @private
*/
exports.ensureArray = (maybeArray) => {
@ -85,9 +86,11 @@ exports.ensureArray = (maybeArray) => {
/**
* Ensures `message.from` is base58 encoded
* @param {object} message
* @param {String} peerId
* @return {object}
*
* @template {{from?:any}} T
* @param {T & {from?:string, receivedFrom:string}} message
* @param {string} [peerId]
* @returns {T & {from?: string, peerId?: string }}
*/
exports.normalizeInRpcMessage = (message, peerId) => {
const m = Object.assign({}, message)
@ -101,8 +104,10 @@ exports.normalizeInRpcMessage = (message, peerId) => {
}
/**
* @param {object} message
* @return {object}
* @template {{from?:any, data?:any}} T
*
* @param {T} message
* @returns {T & {from?: Uint8Array, data?: Uint8Array}}
*/
exports.normalizeOutRpcMessage = (message) => {
const m = Object.assign({}, message)

21
src/record/README.md
View File

@ -36,15 +36,30 @@ const fromString = require('uint8arrays/from-string')
const ENVELOPE_DOMAIN_PEER_RECORD = 'libp2p-peer-record'
const ENVELOPE_PAYLOAD_TYPE_PEER_RECORD = fromString('0301', 'hex')
class PeerRecord extends Record {
/**
* @implements {import('libp2p-interfaces/src/record/types').Record}
*/
class PeerRecord {
constructor (peerId, multiaddrs, seqNumber) {
super (ENVELOPE_DOMAIN_PEER_RECORD, ENVELOPE_PAYLOAD_TYPE_PEER_RECORD)
this.domain = ENVELOPE_DOMAIN_PEER_RECORD
this.codec = ENVELOPE_PAYLOAD_TYPE_PEER_RECORD
}
/**
* Marshal a record to be used in an envelope.
*
* @returns {Uint8Array}
*/
marshal () {
// Implement and return using Protobuf
}
/**
* Returns true if `this` record equals the `other`.
*
* @param {PeerRecord} other
* @returns {other is Record}
*/
equals (other) {
// Verify
}
@ -73,4 +88,4 @@ Verifies if the other Record is identical to this one.
- other is a `Record` to compare with the current instance.
**Returns**
- `boolean`
- `other is Record`

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

@ -1,23 +0,0 @@
export = Record;
/**
* Record is the base implementation of a record that can be used as the payload of a libp2p envelope.
*/
declare class Record {
/**
* @constructor
* @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.
*/
marshal(): void;
/**
* Verifies if the other provided Record is identical to this one.
* @param {Record} other
*/
equals(other: Record): void;
}

35
src/record/index.js
View File

@ -1,35 +0,0 @@
'use strict'
const errcode = require('err-code')
/**
* Record is the base implementation of a record that can be used as the payload of a libp2p envelope.
*/
class Record {
/**
* @constructor
* @param {String} domain signature domain
* @param {Uint8Array} codec identifier of the type of record
*/
constructor (domain, codec) {
this.domain = domain
this.codec = codec
}
/**
* Marshal a record to be used in an envelope.
*/
marshal () {
throw errcode(new Error('marshal must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
}
/**
* Verifies if the other provided Record is identical to this one.
* @param {Record} other
*/
equals (other) {
throw errcode(new Error('equals must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
}
}
module.exports = Record

21
src/record/types.ts Normal file
View File

@ -0,0 +1,21 @@
/**
* Record is the base implementation of a record that can be used as the payload of a libp2p envelope.
*/
export interface Record {
/**
* signature domain.
*/
domain: string;
/**
* identifier of the type of record
*/
codec: Uint8Array;
/**
* Marshal a record to be used in an envelope.
*/
marshal(): Uint8Array;
/**
* Verifies if the other provided Record is identical to this one.
*/
equals(other: unknown): boolean
}

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

@ -5,7 +5,7 @@ const chai = require('chai')
chai.use(require('chai-checkmark'))
const { expect } = chai
const pair = require('it-pair/duplex')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const { collect, map, consume } = require('streaming-iterables')
function close (stream) {
@ -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) {

6
src/stream-muxer/tests/close-test.js
View File

@ -3,12 +3,12 @@
'use strict'
const pair = require('it-pair/duplex')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const { consume } = require('streaming-iterables')
const Tcp = require('libp2p-tcp')
const multiaddr = require('multiaddr')
const abortable = require('abortable-iterator')
const AbortController = require('abort-controller')
const { source: abortable } = require('abortable-iterator')
const AbortController = require('abort-controller').default
const uint8arrayFromString = require('uint8arrays/from-string')
const mh = multiaddr('/ip4/127.0.0.1/tcp/0')

10
src/stream-muxer/tests/spawner.js
View File

@ -2,8 +2,9 @@
const { expect } = require('chai')
const pair = require('it-pair/duplex')
const pipe = require('it-pipe')
const pLimit = require('p-limit')
const { pipe } = require('it-pipe')
const pLimit = require('p-limit').default
const { collect, tap, consume } = require('streaming-iterables')
module.exports = async (Muxer, nStreams, nMsg, limit) => {
@ -61,8 +62,11 @@ module.exports = async (Muxer, nStreams, nMsg, limit) => {
}
function marker (n) {
/** @type {Function} */
let check
let i = 0
/** @type {Promise<void>} */
const done = new Promise((resolve, reject) => {
check = err => {
i++
@ -78,5 +82,7 @@ function marker (n) {
}
}
})
// @ts-ignore - TS can't see that assignement occured
return { check, done }
}

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

@ -0,0 +1,51 @@
import BufferList from 'bl'
export interface MuxerFactory {
new (options: MuxerOptions): Muxer;
multicodec: string;
}
/**
* A libp2p stream muxer
*/
export interface Muxer {
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.
*/
newStream (name?: string): MuxedStream;
/**
* A function called when receiving a new stream from the remote.
*/
onStream (stream: MuxedStream): void;
/**
* A function called when a stream ends.
*/
onStreamEnd (stream: MuxedStream): void;
}
export type MuxerOptions = {
onStream: (stream: MuxedStream) => void;
onStreamEnd: (stream: MuxedStream) => void;
maxMsgSize?: number;
}
export type MuxedTimeline = {
open: number;
close?: number;
}
export interface MuxedStream extends AsyncIterable<Uint8Array | BufferList> {
close: () => void;
abort: () => void;
reset: () => void;
sink: Sink;
source: () => AsyncIterable<Uint8Array | BufferList>;
timeline: MuxedTimeline;
id: string;
}
export type Sink = (source: Uint8Array) => Promise<Uint8Array>;

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

@ -1,42 +0,0 @@
declare const _exports: Topology;
export = _exports;
declare class Topology {
/**
* @param {Object} props
* @param {number} props.min minimum needed connections (default: 0)
* @param {number} props.max maximum needed connections (default: Infinity)
* @param {Object} [props.handlers]
* @param {function} [props.handlers.onConnect] protocol "onConnect" handler
* @param {function} [props.handlers.onDisconnect] protocol "onDisconnect" handler
* @constructor
*/
constructor({ min, max, handlers }: {
min: number;
max: number;
handlers?: {
onConnect?: Function;
onDisconnect?: Function;
};
});
min: number;
max: number;
_onConnect: Function;
_onDisconnect: Function;
/**
* Set of peers that support the protocol.
* @type {Set<string>}
*/
peers: Set<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;
}

60
src/topology/index.js
View File

@ -1,17 +1,28 @@
'use strict'
const withIs = require('class-is')
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: Connection) => void} [onConnect] - protocol "onConnect" handler
* @property {(peerId: PeerId, error?:Error) => void} [onDisconnect] - protocol "onDisconnect" handler
*
* @typedef {import('../connection/connection')} Connection
*/
class Topology {
/**
* @param {Object} props
* @param {number} props.min minimum needed connections (default: 0)
* @param {number} props.max maximum needed connections (default: Infinity)
* @param {Object} [props.handlers]
* @param {function} [props.handlers.onConnect] protocol "onConnect" handler
* @param {function} [props.handlers.onDisconnect] protocol "onDisconnect" handler
* @constructor
* @param {Options} options
*/
constructor ({
min = 0,
@ -27,22 +38,37 @@ class Topology {
/**
* Set of peers that support the protocol.
*
* @type {Set<string>}
*/
this.peers = new Set()
}
set registrar (registrar) {
get [Symbol.toStringTag] () {
return 'Topology'
}
get [topologySymbol] () {
return true
}
/**
* Checks if the given value is a Topology instance.
*
* @param {any} other
* @returns {other is Topology}
*/
static isTopology (other) {
return Boolean(other && other[topologySymbol])
}
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}
*/
@ -51,8 +77,4 @@ class Topology {
}
}
/**
* @module
* @type {Topology}
*/
module.exports = withIs(Topology, { className: 'Topology', symbolName: '@libp2p/js-interfaces/topology' })
module.exports = Topology

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

@ -1,52 +0,0 @@
declare const _exports: MulticodecTopology;
export = _exports;
declare class MulticodecTopology {
/**
* @param {Object} props
* @param {number} props.min minimum needed connections (default: 0)
* @param {number} props.max maximum needed connections (default: Infinity)
* @param {Array<string>} props.multicodecs protocol multicodecs
* @param {Object} props.handlers
* @param {function} props.handlers.onConnect protocol "onConnect" handler
* @param {function} props.handlers.onDisconnect protocol "onDisconnect" handler
* @constructor
*/
constructor({ min, max, multicodecs, handlers }: {
min: number;
max: number;
multicodecs: string[];
handlers: {
onConnect: Function;
onDisconnect: Function;
};
});
multicodecs: string[];
_registrar: any;
/**
* Check if a new peer support the multicodecs for this topology.
* @param {Object} props
* @param {PeerId} props.peerId
* @param {Array<string>} props.protocols
*/
_onProtocolChange({ peerId, protocols }: {
peerId: any;
protocols: string[];
}): void;
/**
* Verify if a new connected peer has a topology multicodec and call _onConnect.
* @param {Connection} connection
* @returns {void}
*/
_onPeerConnect(connection: any): void;
set registrar(arg: any);
/**
* Update topology.
* @param {Array<{id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string>}>} peerDataIterable
* @returns {void}
*/
_updatePeers(peerDataIterable: {
id: any;
multiaddrs: any[];
protocols: string[];
}[]): void;
}

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

@ -1,19 +1,11 @@
'use strict'
const withIs = require('class-is')
const Topology = require('./index')
const multicodecTopologySymbol = Symbol.for('@libp2p/js-interfaces/topology/multicodec-topology')
class MulticodecTopology extends Topology {
/**
* @param {Object} props
* @param {number} props.min minimum needed connections (default: 0)
* @param {number} props.max maximum needed connections (default: Infinity)
* @param {Array<string>} props.multicodecs protocol multicodecs
* @param {Object} props.handlers
* @param {function} props.handlers.onConnect protocol "onConnect" handler
* @param {function} props.handlers.onDisconnect protocol "onDisconnect" handler
* @constructor
* @param {TopologyOptions & MulticodecOptions} props
*/
constructor ({
min,
@ -46,7 +38,25 @@ class MulticodecTopology extends Topology {
this._onPeerConnect = this._onPeerConnect.bind(this)
}
set registrar (registrar) {
get [Symbol.toStringTag] () {
return 'Topology'
}
get [multicodecTopologySymbol] () {
return true
}
/**
* Checks if the given value is a `MulticodecTopology` instance.
*
* @param {any} other
* @returns {other is MulticodecTopology}
*/
static isMulticodecTopology (other) {
return Boolean(other && other[multicodecTopologySymbol])
}
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)
@ -57,6 +67,7 @@ class MulticodecTopology extends Topology {
/**
* Update topology.
*
* @param {Array<{id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string>}>} peerDataIterable
* @returns {void}
*/
@ -77,6 +88,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
@ -102,10 +114,12 @@ class MulticodecTopology extends Topology {
/**
* Verify if a new connected peer has a topology multicodec and call _onConnect.
*
* @param {Connection} connection
* @returns {void}
*/
_onPeerConnect (connection) {
// @ts-ignore - remotePeer does not existist on Connection
const peerId = connection.remotePeer
const protocols = this._registrar.peerStore.protoBook.get(peerId)
@ -121,7 +135,13 @@ class MulticodecTopology extends Topology {
}
/**
* @module
* @type {MulticodecTopology}
* @typedef {import('peer-id')} PeerId
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('../connection/connection')} Connection
* @typedef {import('.').Options} TopologyOptions
* @typedef {Object} MulticodecOptions
* @property {string[]} multicodecs - protocol multicodecs
* @property {Required<Handlers>} handlers
* @typedef {import('.').Handlers} Handlers
*/
module.exports = withIs(MulticodecTopology, { className: 'MulticodecTopology', symbolName: '@libp2p/js-interfaces/topology/multicodec-topology' })
module.exports = MulticodecTopology

6
src/transport/errors.d.ts vendored
View File

@ -1,6 +0,0 @@
export class AbortError extends Error {
static get code(): string;
static get type(): string;
code: string;
type: string;
}

4
src/transport/tests/dial-test.js
View File

@ -9,8 +9,8 @@ chai.use(dirtyChai)
const { isValidTick } = require('./utils')
const goodbye = require('it-goodbye')
const { collect } = require('streaming-iterables')
const pipe = require('it-pipe')
const AbortController = require('abort-controller')
const { pipe } = require('it-pipe')
const AbortController = require('abort-controller').default
const AbortError = require('../errors').AbortError
const sinon = require('sinon')

2
src/transport/tests/listen-test.js
View File

@ -9,7 +9,7 @@ chai.use(dirtyChai)
const sinon = require('sinon')
const pWaitFor = require('p-wait-for')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const uint8arrayFromString = require('uint8arrays/from-string')
const { isValidTick } = require('./utils')

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) {

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

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

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) {

21
tsconfig.json
View File

@ -1,18 +1,9 @@
{
"include": ["src/**/*.js"],
"exclude": ["src/**/tests/*", "src/utils"],
"extends": "./node_modules/aegir/src/config/tsconfig.aegir.json",
"compilerOptions": {
// Tells TypeScript to read JS files, as
// normally they are ignored as source files
"allowJs": true,
// Generate d.ts files
"declaration": true,
// This compiler run should
// only output d.ts files
"emitDeclarationOnly": true,
"esModuleInterop": true,
"rootDir": "./src",
"outDir": "./src"
}
"outDir": "dist"
},
"include": [
"src"
]
}