import { Injectable } from '@angular/core' ; import { webSocket } from 'rxjs/webSocket' ; import { environment } from '../../environments/environment' ; import { Observable , timer , Subject , EMPTY } from 'rxjs' ; import { retryWhen , tap , delayWhen , switchAll , catchError } from 'rxjs/operators' ; export const WS_ENDPOINT = environment . wsEndpoint ; export const RECONNECT_INTERVAL = environment . reconnectInterval ; @ Injectable ( { providedIn : 'root' } ) export class DataService { private socket$ ; private messagesSubject$ = new Subject ( ) ; public messages$ = this . messagesSubject$ . pipe ( switchAll ( ) , catchError ( e => { throw e ; } ) ) ; constructor ( ) { } param cfg public connect ( cfg = { reconnect : false } ) { if ( ! this . socket$ || this . socket$ . closed ) { this . socket$ = this . getNewWebSocket ( ) ; const messages = this . socket$ . pipe ( cfg . reconnect ? this . reconnect : o => o , tap ( { error : error => console . log ( error ) , } ) , catchError ( _ => EMPTY ) ) ; this . messagesSubject$ . next ( messages ) ; } } param observable private reconnect ( observable ) { return observable . pipe ( retryWhen ( errors => errors . pipe ( tap ( val => console . log ( '[Data Service] Try to reconnect' , val ) ) , delayWhen ( _ => timer ( RECONNECT_INTERVAL ) ) ) ) ) ; } close ( ) { this . socket$ . complete ( ) ; this . socket$ = undefined ; } sendMessage ( msg ) { this . socket$ . next ( msg ) ; } /**
     * Return a custom WebSocket subject which reconnects after failure
     */ private getNewWebSocket ( ) { return webSocket ( { url : WS_ENDPOINT , openObserver : { next : ( ) => { console . log ( '[DataService]: connection ok' ) ; } } , closeObserver : { next : ( ) => { console . log ( '[DataService]: connection closed' ) ; this . socket$ = undefined ; this . connect ( { reconnect : true } ) ; } } , } ) ; } }