typescript - Återansluta en websocket i Angular och rxjs?

Jag har en ngrx/store (v2.2.2) och rxjs (v5.1.0) baserad applikation som lyssnar på en webbsocket för inkommande data med hjälp av en observable. När jag startar applikationen tar jag emot inkommande data utan problem.

Men efter ett tag (uppdateringar kommer ganska sällan) verkar anslutningen gå förlorad och jag får inga fler inkommande data. Min kod:

Tjänsten

import { Injectable, OnInit } from '@angular/core';
import { Observable } from 'rxjs';

@Injectable()
export class MemberService implements OnInit {

  private websocket: any;
  private destination: string = "wss://notessensei.mybluemix.net/ws/time";

  constructor() { }

  ngOnInit() { }

  listenToTheSocket(): Observable<any> {

    this.websocket = new WebSocket(this.destination);

    this.websocket.onopen = () => {
      console.log("WebService Connected to " + this.destination);
    }

    return Observable.create(observer => {
      this.websocket.onmessage = (evt) => {
        observer.next(evt);
      };
    })
      .map(res => res.data)
      .share();
  }
}

Abonnenten

  export class AppComponent implements OnInit {

  constructor(/*private store: Store<fromRoot.State>,*/ private memberService: MemberService) {}

  ngOnInit() {
    this.memberService.listenToTheSocket().subscribe((result: any) => {
      try {
        console.log(result);
        // const member: any = JSON.parse(result);
        // this.store.dispatch(new MemberActions.AddMember(member));
      } catch (ex) {
        console.log(JSON.stringify(ex));
      }
    })
  }
}

Vad behöver jag göra för att återansluta webbsocketen när den timear ut, så att observabeln fortsätter att sända inkommande värden?

Jag har tittat på några frågor och svar här, här och här, och det verkade inte ta upp den här frågan (på ett sätt jag kunde förstå).

Observera: websocketen på wss://notessensei.mybluemix.net/ws/time är live och skickar en tidsstämpel en gång i minuten (om man vill testa det).

Råd uppskattas mycket!

Svar #1

Faktiskt finns det nu ett WebsocketSubject i rxjs!

 import { webSocket } from 'rxjs/webSocket' // for RxJS 6, for v5 use Observable.webSocket

 let subject = webSocket('ws://localhost:8081');
 subject.subscribe(
    (msg) => console.log('message received: ' + msg),
    (err) => console.log(err),
    () => console.log('complete')
  );
 subject.next(JSON.stringify({ op: 'hello' }));

Det hanterar återanslutning när du prenumererar på en bruten anslutning igen. Så till exempel skriv detta för att återansluta:

subject.retry().subscribe(...)

Se dokumentationen för mer information. Tyvärr visar inte sökrutan metoden, men du hittar den här:

http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#static-method-webSocket

den #-navigeringen fungerar inte i min webbläsare, så sök efter "webSocket" på den sidan.

Källa: http://reactivex.io/rxjs/file/es6/observable/dom/WebSocketSubject.js.html#lineNumber15

Svar #2

För rxjs 6-implementering

import { webSocket } from 'rxjs/webSocket'
import { retry, RetryConfig } from "rxjs/operators";

const retryConfig: RetryConfig = {
  delay: 3000,
};

let subject = webSocket('ws://localhost:8081');
subject.pipe(
   retry(retryConfig) //support auto reconnect
).subscribe(...)

Relaterade artiklar