src/shared/services/WebSocketService.ts

   1 import { Observable } from "rxjs";
   2 import { share } from "rxjs/operators";
   3 import {
   4   ExponentialBackoff,
   5   LRUBuffer,
   6   Websocket as WS,
   7   WebsocketBuilder,
   8 } from "websocket-ts";
   9 import { getWsUri } from "../env";
  10 import { isBrowser } from "../utils";
  11
  12 export class WebSocketService {
  13   private static _instance: WebSocketService;
  14   private ws: WS;
  15   public subject: Observable<any>;
  16
  17   private constructor() {
  18     let firstConnect = true;
  19
  20     this.subject = new Observable((obs: any) => {
  21       this.ws = new WebsocketBuilder(getWsUri())
  22         .onMessage((_i, e) => {
  23           try {
  24             obs.next(JSON.parse(e.data.toString()));
  25           } catch (err) {
  26             console.error(err);
  27           }
  28         })
  29         .onOpen(() => {
  30           console.log(`Connected to ${getWsUri()}`);
  31
  32           if (!firstConnect) {
  33             let res = {
  34               reconnect: true,
  35             };
  36             obs.next(res);
  37           }
  38           firstConnect = false;
  39         })
  40         .onRetry(() => {
  41           console.log("Retrying websocket connection...");
  42         })
  43         .onClose(() => {
  44           console.error("Websocket closed.");
  45         })
  46         .withBackoff(new ExponentialBackoff(100, 7))
  47         .withBuffer(new LRUBuffer(1000))
  48         .build();
  49     }).pipe(share());
  50
  51     if (isBrowser()) {
  52       window.onunload = () => {
  53         this.ws.close();
  54
  55         // Clears out scroll positions.
  56         sessionStorage.clear();
  57       };
  58     }
  59   }
  60
  61   public send(data: string) {
  62     this.ws.send(data);
  63   }
  64
  65   public static get Instance() {
  66     return this._instance || (this._instance = new this());
  67   }
  68 }