Tengo una aplicación de chat que consta de varios componentes (por ejemplo, un componente contiene una lista de grupos, otro contiene mensajes, etc.), y cuando se abre un componente, se debe cerrar otro.
Así que aquí hay un servicio que estoy usando para proporcionar una conexión webSocket:
@Injectable() export class ChatConnectionService { private _connection: Subject<any>; constructor(private window: WindowAbstract) { } public closeConnection(): void { this.connection.complete(); } public send(msg: ChatRequest): void { this.connection.next(msg); } public messages(): Subject<any> { return this.connection; } private get connection(): Subject<any> { if (!this._connection || this._connection.closed) { this._connection = webSocket(`https/myapp/chat`); } return this._connection; } } Y aquí hay un ejemplo de uso en some.component.ts :
private messages$: Subscription; ngOnInit(): void { this.messages$ = this.chatConnectionService.messages() .subscribe(m => this.handleInboundMessage(m)); } ngOnDestroy() { this.messages$.unsubscribe(); }Ahora, el problema es que cuando cambio entre componentes, una vez que se da de baja cuando se destruye un componente anterior, una nueva suscripción no parece funcionar, no envía ni recibe mensajes y tampoco produce ningún error.
Pero cuando elimino la línea de cancelación de suscripción, comienza a funcionar, pero también sigue recibiendo mensajes usando la suscripción anterior, que no es lo que realmente quiero.
Entonces, resumiendo todo: necesito mantener una conexión pero tener la posibilidad de que varios componentes se suscriban y cancelen la suscripción según sea necesario, ¿es posible lograrlo con RxJs? O que estoy haciendo mal? Cualquier sugerencia será apreciada, gracias!
El problema es que está dando a los componentes una referencia directa al WebSocketSubject que se devuelve desde webSocket('https/myapp/chat') . Si invoca el método .unsubscribe() en WebSocketSubject , cierra el canal, por lo que ningún otro componente detecta mensajes.
En lugar de hacer captadores, declare un observable público que se canaliza desde el WebSocketSubject privado.
@Injectable() export class ChatConnectionService { public messages$ = Observable<any>; // Replace 'any' with WS response type. private connection: WebSocketSubject; constructor(private window: WindowAbstract) { this.connection = webSocket(`https/myapp/chat`); this.messages$ = this.connection.asObservable(); } }Su componente ahora cancelará la suscripción del observable, en lugar de invocar WebSocketSubject para cancelar la suscripción del canal.
Actualización Después de una revisión adicional del WebSocketSubject creado por webSocket() , me di cuenta de que .asObservable() no es el mejor caso de uso para su situación. Incluso si sus componentes se dan de baja del observable, el Sujeto sigue recibiendo nuevos datos del canal WS abierto.
this.messages$ = this.connection.multiplex( () => { 'send message to WS server to start' }, () => { 'send message to WS server to stop' }, (message) => true // filter messages by condition ); Puede declarar un observable desde su WebSocketSubject utilizando el método multiplex() . Toma tres parámetros:
Ahora, cada vez que cualquier componente se suscriba a los messages$ , activará el mensaje de inicio, abriendo el canal para los suscriptores. Cuando todos los observables se dan de baja de messages$ , envía la señal de cierre que detiene cualquier mensaje adicional.
Esto ahorra una gran cantidad de repeticiones para mantener un estado singleton para su WS.
Referencia: https://rxjs.dev/api/webSocket/WebSocketSubject
Lo dejaré aquí para alguien que termine en la misma situación.
Necesita al menos una suscripción activa todo el tiempo para que no se cierre la conexión cuando cancela la suscripción.
Así que esto es lo que hice:
@NgModule({ imports: [ ... ], providers: [ { provide: APP_INITIALIZER, multi: true, // subscribing establishes a connections useFactory: (chat: ChatConnectionService) => () => chat.messages$.subscribe(), deps: [ChatConnectionService] } ] }) export class AppModule { } El escollo no tan obvio aquí es que darse de baja de WebSocketSubject y de un Observable , que se creó al suscribirse al sujeto, tiene el mismo efecto: cierra una conexión activa si fue la última suscripción.