Búsqueda de sitios web

typescript - ¿Reconectando un websocket en Angular y rxjs?

Tengo una aplicación basada en ngrx/store (v2.2.2) y rxjs (v5.1.0) que escucha un socket web para recibir datos usando un observable. Cuando inicio la aplicación, recibo datos entrantes sin problemas.

Sin embargo, después de un tiempo (las actualizaciones llegan con bastante poca frecuencia) la conexión parece perderse y ya no recibo más datos entrantes. Mi código:

El servicio

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

El suscriptor

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

¿Qué necesito hacer para reconectar el socket web cuando se agota el tiempo de espera, para que el observable continúe emitiendo valores entrantes?

Eché un vistazo a algunas preguntas y respuestas aquí, aquí y aquí y no parecían abordar esta pregunta (de una manera que pudiera comprender).

Nota: el websocket en wss://notessensei.mybluemix.net/ws/time está activo y emite una marca de tiempo una vez por minuto (por si alguien quiere probarlo).

¡Se agradece mucho el consejo!

Respuesta #1

¡De hecho, ahora hay un WebsocketSubject en 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' }));

Sí maneja la reconexión cuando te vuelves a suscribir a una conexión interrumpida. Así que, por ejemplo, escribe esto para reconectar:

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

Consulta la documentación para más información. Desafortunadamente, el cuadro de búsqueda no muestra el método, pero lo puedes encontrar aquí:

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

esa #-navegación no funciona en mi navegador, así que busca "webSocket" en esa página.

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

Respuesta #2

Para la implementación de rxjs 6

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(...)

Artículos relacionados