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>|null {
    if (!this.socket || this.socket.closed)
      return null;

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

  private getNewWebSocket() {
    return webSocket({
      url: environment.WSS_ENDPOINT
    });
  }

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

    this.socket.next(msg);
  }

  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();
  }
}