-
Notifications
You must be signed in to change notification settings - Fork 59
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add web socket based network info provider
- Loading branch information
Showing
26 changed files
with
955 additions
and
27 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,228 @@ | ||
| /* eslint-disable unicorn/prefer-add-event-listener */ | ||
|
|
||
| import { | ||
| AsyncReturnType, | ||
| Cardano, | ||
| EpochInfo, | ||
| EraSummary, | ||
| HealthCheckResponse, | ||
| NetworkInfoMethods, | ||
| NetworkInfoProvider, | ||
| StakeSummary, | ||
| SupplySummary, | ||
| WSMessage, | ||
| createSlotEpochInfoCalc | ||
| } from '@cardano-sdk/core'; | ||
| import { Logger } from 'ts-log'; | ||
| import { Observable, ReplaySubject, Subject, firstValueFrom } from 'rxjs'; | ||
|
|
||
| import { fromSerializableObject } from '@cardano-sdk/util'; | ||
| import WebSocket from 'isomorphic-ws'; | ||
|
|
||
| const NOT_CONNECTED_ID = 'not-connected'; | ||
|
|
||
| type WSStatus = 'connecting' | 'connected' | 'idle' | 'stop'; | ||
|
|
||
| export type WSHandler = (message: WSMessage) => void; | ||
|
|
||
| export interface WsClientConfiguration { | ||
| /** The interval in seconds between two heartbeat messages. Default 55". */ | ||
| heartbeatInterval?: number; | ||
|
|
||
| /** The interval in seconds after which a request must timeout. Default 60". */ | ||
| requestTimeout?: number; | ||
|
|
||
| /** The WebSocket server URL. */ | ||
| url: URL; | ||
| } | ||
|
|
||
| export interface WsClientDependencies { | ||
| /** The logger. */ | ||
| logger: Logger; | ||
| } | ||
|
|
||
| interface EpochRollover { | ||
| epochInfo: EpochInfo; | ||
| eraSummaries: EraSummary[]; | ||
| ledgerTip: Cardano.Tip; | ||
| lovelaceSupply: SupplySummary; | ||
| protocolParameters: Cardano.ProtocolParameters; | ||
| } | ||
|
|
||
| const isEventError = (error: unknown): error is { error: Error } => | ||
| // eslint-disable-next-line @typescript-eslint/no-explicit-any | ||
| typeof error === 'object' && !!error && (error as any).error instanceof Error; | ||
|
|
||
| export class CardanoWsClient { | ||
| /** The client id, assigned by the server. */ | ||
| clientId = NOT_CONNECTED_ID; | ||
|
|
||
| /** Emits on epoch rollover. */ | ||
| epoch$: Observable<EpochRollover>; | ||
|
|
||
| /** Emits the health state. */ | ||
| health$: Observable<HealthCheckResponse>; | ||
|
|
||
| /** The `Observable` form of `NetworkInfoProvider`. */ | ||
| networkInfo: { | ||
| [m in `${NetworkInfoMethods}$`]: Observable< | ||
| AsyncReturnType<NetworkInfoProvider[m extends `${infer o}$` ? o : never]> | ||
| >; | ||
| }; | ||
|
|
||
| /** WebSocket based `NetworkInfoProvider` implementation. */ | ||
| networkInfoProvider: NetworkInfoProvider; | ||
|
|
||
| private epochSubject$: Subject<EpochRollover>; | ||
| private healthSubject$: ReplaySubject<HealthCheckResponse>; | ||
| private heartbeatInterval: number; | ||
| private heartbeatTimeout: NodeJS.Timeout | undefined; | ||
| private logger: Logger; | ||
| private status: WSStatus = 'idle'; | ||
| private url: URL; | ||
| private ws: WebSocket; | ||
|
|
||
| private networkInfoSubjects = {} as { | ||
| [m in `${NetworkInfoMethods}$`]: ReplaySubject< | ||
| AsyncReturnType<NetworkInfoProvider[m extends `${infer o}$` ? o : never]> | ||
| >; | ||
| }; | ||
|
|
||
| constructor(deps: WsClientDependencies, cfg: WsClientConfiguration) { | ||
| this.epoch$ = this.epochSubject$ = new Subject<EpochRollover>(); | ||
| this.health$ = this.healthSubject$ = new ReplaySubject<HealthCheckResponse>(1); | ||
| this.heartbeatInterval = (cfg.heartbeatInterval || 55) * 1000; | ||
| this.logger = deps.logger; | ||
| this.url = cfg.url; | ||
|
|
||
| this.networkInfoSubjects = { | ||
| eraSummaries$: new ReplaySubject<EraSummary[]>(1), | ||
| genesisParameters$: new ReplaySubject<Cardano.CompactGenesis>(1), | ||
| ledgerTip$: new ReplaySubject<Cardano.Tip>(1), | ||
| lovelaceSupply$: new ReplaySubject<SupplySummary>(1), | ||
| protocolParameters$: new ReplaySubject<Cardano.ProtocolParameters>(1), | ||
| stake$: new ReplaySubject<StakeSummary>(1) | ||
| }; | ||
| this.networkInfo = this.networkInfoSubjects; | ||
|
|
||
| this.networkInfoProvider = { | ||
| eraSummaries: () => firstValueFrom(this.networkInfo.eraSummaries$), | ||
| genesisParameters: () => firstValueFrom(this.networkInfo.genesisParameters$), | ||
| healthCheck: () => firstValueFrom(this.health$), | ||
| ledgerTip: () => firstValueFrom(this.networkInfo.ledgerTip$), | ||
| lovelaceSupply: () => firstValueFrom(this.networkInfo.lovelaceSupply$), | ||
| protocolParameters: () => firstValueFrom(this.networkInfo.protocolParameters$), | ||
| stake: () => firstValueFrom(this.networkInfo.stake$) | ||
| }; | ||
|
|
||
| this.connect(); | ||
| } | ||
|
|
||
| private connect() { | ||
| this.status = 'connecting'; | ||
|
|
||
| const ws = new WebSocket(this.url); | ||
|
|
||
| // eslint-disable-next-line sonarjs/cognitive-complexity, complexity | ||
| ws.onmessage = (event) => { | ||
| try { | ||
| if (typeof event.data !== 'string') throw new Error('Unexpected data from WebSocket '); | ||
|
|
||
| const message = fromSerializableObject<WSMessage>(JSON.parse(event.data)); | ||
| const { clientId, networkInfo } = message; | ||
|
|
||
| if (clientId) this.logger.info(`Connected with clientId ${(this.clientId = clientId)}`); | ||
|
|
||
| if (networkInfo) { | ||
| const { eraSummaries, genesisParameters, ledgerTip, lovelaceSupply, protocolParameters, stake } = networkInfo; | ||
|
|
||
| if (eraSummaries) this.networkInfoSubjects.eraSummaries$.next(eraSummaries); | ||
| if (genesisParameters) this.networkInfoSubjects.genesisParameters$.next(genesisParameters); | ||
| if (lovelaceSupply) this.networkInfoSubjects.lovelaceSupply$.next(lovelaceSupply); | ||
| if (protocolParameters) this.networkInfoSubjects.protocolParameters$.next(protocolParameters); | ||
| if (stake) this.networkInfoSubjects.stake$.next(stake); | ||
|
|
||
| // Emit ledgerTip as last one | ||
| if (ledgerTip) this.networkInfoSubjects.ledgerTip$.next(ledgerTip); | ||
|
|
||
| // If it is an epoch rollover, emit it | ||
| if (eraSummaries && ledgerTip && lovelaceSupply && protocolParameters && !clientId) { | ||
| const epochInfo = createSlotEpochInfoCalc(eraSummaries)(ledgerTip.slot); | ||
|
|
||
| this.epochSubject$.next({ epochInfo, eraSummaries, ledgerTip, lovelaceSupply, protocolParameters }); | ||
| } | ||
| } | ||
| } catch (error) { | ||
| this.logger.error(error, 'While parsing message', event.data, this.clientId); | ||
| } | ||
| }; | ||
|
|
||
| ws.onclose = () => { | ||
| this.logger.info('WebSocket client connection closed', this.clientId); | ||
|
|
||
| if (this.heartbeatTimeout) { | ||
| clearInterval(this.heartbeatTimeout); | ||
| this.heartbeatTimeout = undefined; | ||
| } | ||
|
|
||
| this.clientId = NOT_CONNECTED_ID; | ||
| this.retry(); | ||
| }; | ||
|
|
||
| ws.onerror = (error: unknown) => { | ||
| const err = isEventError(error) ? error.error : new Error(`Unknown error: ${JSON.stringify(error)}`); | ||
|
|
||
| this.logger.error(err, 'Async error from WebSocket client', this.clientId); | ||
| ws.close(); | ||
| this.healthSubject$.next({ ok: false, reason: err.message }); | ||
| }; | ||
|
|
||
| ws.onopen = () => { | ||
| this.status = 'connected'; | ||
| this.ws = ws; | ||
| this.heartbeat(); | ||
| this.healthSubject$.next({ ok: true }); | ||
| }; | ||
| } | ||
|
|
||
| private heartbeat() { | ||
| if (this.heartbeatTimeout) clearInterval(this.heartbeatTimeout); | ||
|
|
||
| this.heartbeatTimeout = setTimeout(() => { | ||
| try { | ||
| this.request({}); | ||
| } catch (error) { | ||
| this.logger.error(error, 'Error while refreshing heartbeat', this.clientId); | ||
| } | ||
| }, this.heartbeatInterval); | ||
| this.heartbeatTimeout.unref(); | ||
| } | ||
|
|
||
| private retry() { | ||
| if (this.status !== 'stop') { | ||
| this.status = 'idle'; | ||
| setTimeout(() => this.connect(), 10_000).unref(); | ||
| } | ||
| } | ||
|
|
||
| /** Closes the WebSocket connection. */ | ||
| close() { | ||
| this.status = 'stop'; | ||
| this.ws.close(); | ||
| } | ||
|
|
||
| /** | ||
| * Sends a request through WS to server. | ||
| * | ||
| * @param request the request. | ||
| * @returns `true` is sent, otherwise `false`. | ||
| */ | ||
| private request(request: WSMessage) { | ||
| if (this.status !== 'connected') return false; | ||
|
|
||
| this.ws!.send(JSON.stringify(request)); | ||
| this.heartbeat(); | ||
|
|
||
| return true; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.