Line data    Source code

       1              : import type { Tick, TickFeed, TickFeedStatus } from './tick_feed.js';
       2              : 
       3              : /// The longest wait before a reconnect
       4            1 : const MAX_RECONNECT_DELAY_MS = 30 * 1000;
       5              : 
       6              : /// What a message of a market data stream holds for the feed
       7              : export type FeedMessage =
       8              :     /// A trade
       9              :     | { tick: Tick }
      10              :     /// An error that the stream reports
      11              :     | { error: string }
      12              :     /// Anything else, e.g. the confirmation of a subscription or a quote that the next trade carries
      13              :     | null;
      14              : 
      15              : /// The protocol of a market data stream
      16              : export interface FeedProtocol {
      17              :     /// The url of the stream
      18              :     url: string;
      19              :     /// The message to send once the socket is open, e.g. a subscription, null if the url subscribes
      20              :     subscription(): string | null;
      21              :     /// Read a text message of the stream
      22              :     read(data: string): FeedMessage;
      23              : }
      24              : 
      25              : /// The options of a WebSocket feed
      26              : export interface WebSocketFeedOptions {
      27              :     /// The time between two deliveries
      28              :     emitIntervalMs?: number;
      29              :     /// Create the socket, for tests
      30              :     createSocket?: (url: string) => WebSocket;
      31              : }
      32              : 
      33              : /// The trades of a market data stream over a WebSocket, delivered in batches.
      34              : /// The feed reconnects with an increasing delay when the socket closes.
      35            1 : export class WebSocketTickFeed implements TickFeed {
      36              :     /// The socket
      37            1 :     private socket: WebSocket | null = null;
      38              :     /// The ticks since the last delivery
      39            1 :     private pending: Tick[] = [];
      40              :     /// The timer of the deliveries
      41            1 :     private emitTimer: ReturnType<typeof setInterval> | null = null;
      42              :     /// The timer of the next reconnect
      43            1 :     private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
      44              :     /// The delay of the next reconnect
      45            1 :     private reconnectDelayMs = 1000;
      46              :     /// Is the feed stopped?
      47            1 :     private stopped = true;
      48              : 
      49            1 :     private readonly emitIntervalMs: number;
      50            1 :     private readonly createSocket: (url: string) => WebSocket;
      51              : 
      52            1 :     constructor(private readonly protocol: FeedProtocol, options: WebSocketFeedOptions = {}) {
      53            1 :         this.emitIntervalMs = options.emitIntervalMs ?? 100;
      54            1 :         this.createSocket = options.createSocket ?? (url => new WebSocket(url));
      55            1 :     }
      56              : 
      57            1 :     start(emit: (ticks: Tick[]) => void, onStatus: (status: TickFeedStatus) => void): void {
      58            1 :         this.stop();
      59            1 :         this.stopped = false;
      60            1 :         this.emitTimer = setInterval(() => {
      61            1 :             if (this.pending.length === 0) return;
      62            1 :             const ticks = this.pending;
      63            1 :             this.pending = [];
      64            1 :             emit(ticks);
      65            1 :         }, this.emitIntervalMs);
      66              :         // Every trade would report the feed as live again, only changes are reported
      67            1 :         let reported: string | null = null;
      68            1 :         this.connect(status => {
      69            1 :             const key = `${status.state}:${status.detail ?? ''}`;
      70            1 :             if (key === reported) return;
      71            1 :             reported = key;
      72            1 :             onStatus(status);
      73            1 :         });
      74            1 :     }
      75              : 
      76            1 :     private connect(onStatus: (status: TickFeedStatus) => void): void {
      77            1 :         onStatus({ state: 'connecting' });
      78            1 :         const socket = this.createSocket(this.protocol.url);
      79            1 :         this.socket = socket;
      80            1 :         socket.onopen = () => {
      81            1 :             const subscription = this.protocol.subscription();
      82            1 :             if (subscription != null) socket.send(subscription);
      83            1 :         };
      84            1 :         socket.onmessage = (event: MessageEvent) => {
      85            1 :             if (typeof event.data !== 'string') return;
      86            1 :             const message = this.protocol.read(event.data);
      87            1 :             if (message == null) return;
      88            1 :             if ('error' in message) {
      89            0 :                 onStatus({ state: 'failed', detail: message.error });
      90            0 :                 return;
      91            0 :             }
      92            1 :             if (this.pending.length === 0) this.reconnectDelayMs = 1000;
      93            1 :             this.pending.push(message.tick);
      94            1 :             onStatus({ state: 'live' });
      95            1 :         };
      96            1 :         socket.onclose = () => {
      97            1 :             if (this.stopped || this.socket !== socket) return;
      98            1 :             this.socket = null;
      99            1 :             const delay = this.reconnectDelayMs;
     100            1 :             this.reconnectDelayMs = Math.min(delay * 2, MAX_RECONNECT_DELAY_MS);
     101            1 :             onStatus({ state: 'connecting', detail: `Reconnecting in ${Math.round(delay / 1000)}s` });
     102            1 :             this.reconnectTimer = setTimeout(() => {
     103            1 :                 this.reconnectTimer = null;
     104            1 :                 if (!this.stopped) this.connect(onStatus);
     105            1 :             }, delay);
     106            1 :         };
     107            1 :     }
     108              : 
     109            1 :     stop(): void {
     110            1 :         this.stopped = true;
     111            1 :         if (this.emitTimer != null) clearInterval(this.emitTimer);
     112            1 :         if (this.reconnectTimer != null) clearTimeout(this.reconnectTimer);
     113            1 :         this.emitTimer = null;
     114            1 :         this.reconnectTimer = null;
     115            1 :         const socket = this.socket;
     116            1 :         this.socket = null;
     117            1 :         socket?.close();
     118            1 :         this.pending = [];
     119            1 :     }
     120            1 : }
     121              : 
     122              : /// A number of a stream, the streams send numbers as strings
     123            1 : export function parseFeedNumber(value: unknown): number | null {
     124            1 :     if (typeof value !== 'string' && typeof value !== 'number') return null;
     125            1 :     const number = Number(value);
     126            1 :     return Number.isFinite(number) ? number : null;
     127            1 : }