web-socket.service.ts

import { Injectable } from '@angular/core';
import { Subject } from 'rxjs';
import {
  // eslint-disable-next-line @typescript-eslint/no-unused-vars
  KLINE_CRYPTOS,
  KLINE_CRYPTOS_BTC,
  // eslint-disable-next-line @typescript-eslint/no-unused-vars
  KLINE_CRYPTOS_QUICK,
} from '../../../../../data/assets.data';
import { LoggingService } from '../logging/logging.service';
import { STRATEGY_INTERVALS, STRATEGY_SYMBOLS } from '../../../../../data/strategy-emac.data';

@Injectable({
  providedIn: 'root',
})
export class WebSocketService {
  //================================================================= PROPERTIES
  private socket: WebSocket | undefined;
  private klineDataRequest: string = KLINE_CRYPTOS_BTC;
  private baseUrl: string = 'wss://stream.binance.com:9443/stream?streams=';
  private symbols = STRATEGY_SYMBOLS;
  private intervals = STRATEGY_INTERVALS;
  private url: string;
  private messageSubject: Subject<string> = new Subject<string>();

  //================================================================ CONSTRUCTOR
  constructor(private loggingService: LoggingService) {}

  //================================================================= CONNECTION
  connect(): void {
    this.url = this.getWebsocketUrl(this.baseUrl, this.symbols, this.intervals);
    this.socket = new WebSocket(this.url);

    this.socket.addEventListener('open', (event) => {
      this.loggingService.log(event.type.toUpperCase());
      this.loggingService.info(this.klineDataRequest);
    });

    /* Websocket data arrives here. */
    this.socket.addEventListener('message', (event) => {
      this.messageSubject.next(event.data);
    });

    this.socket.addEventListener('close', (event) => {
      console.log(event);
    });

    this.socket.addEventListener('error', (event) => {
      console.log(event.type);
    });
  }

  disconnect(): void {
    if (this.socket) {
      this.socket.close();
    }
  }

  //========================================================= WEB SOCKET METHODS
  send(data: unknown): void {
    if (this.socket?.readyState === WebSocket.OPEN) {
      this.socket.send(JSON.stringify(data));
    } else {
      console.error('WebSocket is not open. Unable to send message');
    }
  }

  sendMessageClient() {
    if (this.socket?.readyState === WebSocket.OPEN) {
      this.socket.send('hello from client');
    } else {
      console.error('WebSocket is not open. Unable to send message');
    }
  }

  subscribeToSymbols(symbols: string[], intervals: string[]): void {
    const params: string[] = [];
    symbols.forEach((symbol) => {
      intervals.forEach((interval) => {
        params.push(symbol + '@kline_' + interval);
      });
    });
    const request = { method: 'SUBSCRIBE', params: params, id: 1 };
    console.log(request);
    this.send(request);
  }

  unsubscribeFromSymbol(symbol: string): void {
    this.send({ method: 'UNSUBSCRIBE', symbol });
  }

  /**
   * Returns an observable that emits messages received from the WebSocket.
   *
   * @returns An observable of messages.
   */
  getMessages() {
    return this.messageSubject.asObservable();
  }

  //============================================================ PRIVATE METHODS
  /** Examples
   * `wss://stream.binance.com:9443/ws/btcusdt@kline_1m`
   * `wss://stream.binance.com:9443/stream?streams=usdtars@kline_1m/btcusdt@kline_1m/...` and so on
   * @param baseUrl - The base WebSocket URL.
   * @param symbols - An array of trading symbols (e.g., ['btcusdt', 'ethusdt']).
   * @param intervals - An array of intervals (e.g., ['1m', '5m']).
   * @returns The complete WebSocket URL as a string.
   */
  getWebsocketUrl(
    baseUrl: string,
    symbols: string[],
    intervals: string[]
  ): string {
    let streamName = '';
    symbols.forEach((symbol) => {
      intervals.forEach((interval) => {
        streamName = streamName.concat(symbol, '@kline_', interval, '/');
      });
    });

    // Conditionally remove the last '/'
    if (streamName.endsWith('/')) {
      streamName = streamName.slice(0, -1);
    }

    return baseUrl + streamName;
  }
}