import { logger } from '@libp2p/logger' import mergeOptions from 'merge-options' // @ts-expect-error retimer does not have types import retimer from 'retimer' import all from 'it-all' import { pipe } from 'it-pipe' import filter from 'it-filter' import sort from 'it-sort' import type { Startable } from '@libp2p/interfaces/startable' import type { Components } from '@libp2p/components' const log = logger('libp2p:connection-manager:auto-dialler') export interface AutoDiallerInit { /** * Should preemptively guarantee connections are above the low watermark */ enabled?: boolean /** * The minimum number of connections to avoid pruning */ minConnections?: number /** * How often, in milliseconds, it should preemptively guarantee connections are above the low watermark */ autoDialInterval?: number } const defaultOptions: Partial = { enabled: true, minConnections: 0, autoDialInterval: 10000 } export class AutoDialler implements Startable { private readonly components: Components private readonly options: Required private running: boolean private autoDialTimeout?: ReturnType /** * Proactively tries to connect to known peers stored in the PeerStore. * It will keep the number of connections below the upper limit and sort * the peers to connect based on wether we know their keys and protocols. */ constructor (components: Components, init: AutoDiallerInit) { this.components = components this.options = mergeOptions.call({ ignoreUndefined: true }, defaultOptions, init) this.running = false this._autoDial = this._autoDial.bind(this) log('options: %j', this.options) } isStarted () { return this.running } /** * Starts the auto dialer */ async start () { if (!this.options.enabled) { log('not enabled') return } this.running = true void this._autoDial().catch(err => { log.error('could start autodial', err) }) log('started') } /** * Stops the auto dialler */ async stop () { if (!this.options.enabled) { log('not enabled') return } this.running = false if (this.autoDialTimeout != null) { this.autoDialTimeout.clear() } log('stopped') } async _autoDial () { if (this.autoDialTimeout != null) { this.autoDialTimeout.clear() } const minConnections = this.options.minConnections // Already has enough connections if (this.components.getConnectionManager().getConnections().length >= minConnections) { this.autoDialTimeout = retimer(this._autoDial, this.options.autoDialInterval) return } // Sort peers on whether we know protocols or public keys for them const allPeers = await this.components.getPeerStore().all() const peers = await pipe( // shuffle the peers allPeers.sort(() => Math.random() > 0.5 ? 1 : -1), (source) => filter(source, (peer) => !peer.id.equals(this.components.getPeerId())), (source) => sort(source, (a, b) => { if (b.protocols.length > a.protocols.length) { return 1 } else if (b.id.publicKey != null && a.id.publicKey == null) { return 1 } return -1 }), async (source) => await all(source) ) for (let i = 0; this.running && i < peers.length && this.components.getConnectionManager().getConnections().length < minConnections; i++) { // Connection Manager was stopped during async dial if (!this.running) { return } const peer = peers[i] if (this.components.getConnectionManager().getConnections(peer.id).length === 0) { log('connecting to a peerStore stored peer %p', peer.id) try { await this.components.getConnectionManager().openConnection(peer.id) } catch (err: any) { log.error('could not connect to peerStore stored peer', err) } } } // Connection Manager was stopped if (!this.running) { return } this.autoDialTimeout = retimer(this._autoDial, this.options.autoDialInterval) } }