iot-web-ui/src/app/shared/services/socket.service.ts

import { Injectable } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import {BehaviorSubject, catchError, delayWhen, Observable, retryWhen, Subject, tap, timer} from "rxjs";

@Injectable({
  providedIn: 'root'
})
export class SocketService {
  gateway = 'ws:/socket.aztrace.vn/ws';
  websocket$!: WebSocketSubject<any> ;
  private isConnected: boolean = false;
  private messagesSubject$ = new Subject();
  public messages$ = this.messagesSubject$.asObservable();
  private statusSubject = new BehaviorSubject<boolean>(this.isConnected);
  public status$ = this.statusSubject.asObservable();

  constructor() {
  }

  public connect(cfg: { reconnect: boolean } = {reconnect: false}): void {
    if (!this.websocket$ || this.websocket$.closed) {
      console.log('Trying to open a WebSocket connection…');
      this.websocket$ = this.getNewWebSocket();
      this.websocket$.subscribe((messages) => {
        this.messagesSubject$.next(messages);
      });
    }
  }

  close() {
    this.websocket$.complete();
  }

  sendMessage(msg: any) {
    this.websocket$.next(msg);
  }

  private getNewWebSocket() {
    return webSocket({
      url: this.gateway,
      openObserver: {
        next: () => {
          console.log('Connection ok');
          this.isConnected = true;
          this.statusSubject.next(this.isConnected);
        }
      },
      closeObserver: {
        next: () => {
          console.log('Connection closed');
          this.isConnected = false;
          this.connect({reconnect: true});
          this.statusSubject.next(this.isConnected);
        }
      },
    });
  }
}