rxjs-websockets on angular 4, handling reconnections

Question

I'm currently trying to user rxjs-websockets with angular 4 (https://github.com/ohjames/rxjs-websockets).

Here is my service:

import { Injectable } from '@angular/core';
import { QueueingSubject } from 'queueing-subject'
import { Observable } from 'rxjs/Observable'
import websocketConnect from 'rxjs-websockets'
import { environment } from 'environments/environment';

@Injectable()
export class WebsocketJWTService {
    private inputStream: QueueingSubject < any >
    public messages: Observable < any >
    SERVER: string = environment["SERVER_ADDRESS"];
    constructor(private modal: Modal, private SS: SharedService, private userManager: UserManagerService) {}
    public connect() {
        let SERVER = this.SERVER.substring(8);
        if (this.messages)
            return;
        let temp = websocketConnect(
            "wss://" + SERVER + "/ws/media?token=" + this.userManager.getToken(),
            this.inputStream = new QueueingSubject < any > ()
        )
        this.messages = temp.messages.share()
    }
    public send(message: any): void {
        message['WSTOKEN'] = this.userManager.getToken();
        this.inputStream.next(message)
    }
}

and I'm calling it like this on my components:

constructor(private socket:WebsocketJWTService) {}
ngOnInit() {
    this.socket.connect();
    this.socketSubscription = this.socket.messages.subscribe(message => this.socketMessage(message));
}

Now, I'm trying to handle reconnection so I look the connection status and call the connect function again if the number of connected socket drop to 0:

temp.connectionStatus.subscribe(numberConnected => {
    if (this.connectedNumber > numberConnected) {
        console.log("disconnected");
        setTimeout(() => {
            delete this.messages;
            this.connect();
            this.messages.subscribe();
        }, 5000)
    }
    this.connectedNumber = numberConnected;
})

I'm correctly seeing my socket reconnecting when it drops but my subscription stop working after the reconnection. I'm seeing the socket events beeing sent by the new socket connection but my angular code doesn't seem to be notificated.

Do someone know how I can reconnect to the websocket and keeping my subscription working?

Thank you for you time

Admin · Answer

The link you pasted has a section "Reconnecting on failure". Did you read that, because that's how you do it and it works.

Your own attempt is not going to work, because that is not how rxjs works. subscribe() returns a new subscription and in your reconnect code you aren't doing anything with that subscription. Your old subscription is your old one, after that fails it's not usable anymore.

I'd start with a basic primer on rxjs and read about hot and cold observables before trying to do your own stuff with it, otherwise you can just copy the example in the README of the library that you're using.

_{(Side Note: I am the Library Author).}

user_6581173 · Answer

give something like this a shot. count is just a local variable to keep track of the retries.

this.messages = temp.messages.share()
                            .retryWhen(attempts => {
                                return attempts
                                 .do((error)=>{
                                    return 1})
                                     .mergeMap(error, count)=>{
                                         console.log(`Wait ${count} seconds, then reconnect!`); 
                                  return Observable.timer(count * 1000);}}

rxjs-websockets on angular 4, handling reconnections

2 Answers