hermes-web-angular/src/app/hermes-socket.service.ts

import { OnInit, Injectable } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { catchError, filter, first, timeout } from 'rxjs/operators';
import { environment } from '../environments/environment';
import { Observable, throwError } from 'rxjs';


@Injectable({
  providedIn: 'root'
})
export class HermesSocketService implements OnInit {
  private socket: WebSocketSubject<any> | undefined = undefined

  constructor() { }

  ngOnInit(): void {
  }

  public connect(): void {
    if (!this.socket || this.socket.closed) {
      this.socket = this.getNewWebSocket();
    }
  }

  public first(predicate: (data: any) => boolean): Observable<any> {
    if (!this.socket || this.socket.closed)
      return new Observable().pipe(timeout(3000), catchError((e) => throwError(() => 'No response after 3 seconds.')));

    return this.socket.pipe(timeout(3000), catchError((e) => throwError(() => 'No response after 3 seconds.')), first(predicate));
  }

  private getNewWebSocket(): WebSocketSubject<any> {
    return webSocket({
      url: environment.WSS_ENDPOINT
    });
  }

  public sendMessage(msg: any): void {
    if (!this.socket || this.socket.closed)
      return;

    this.socket.next(msg);
  }

  public get$(): Observable<any>|undefined {
    return this.socket?.asObservable();
  }

  public subscribe(subscriptions: any) {
    if (!this.socket || this.socket.closed)
      return;

    return this.socket.subscribe(subscriptions);
  }

  public close() {
    if (!this.socket || this.socket.closed)
      return;

    this.socket.complete();
  }
}