How to get rxjs retry to work with websocket connection

Question

I'm working on a project which requires the use of a websocket so the server can rapidly push data to the web UI. I'm using Angular v5 with angular-cli. However I'm having a hard time getting the websocket connection to retry when it fails.

For those interested my project in its current state can be found here.

I followed this tutorial but haven't really been able to get to grips with the rxjs principles. I've been doing a lot of trial and error get where I am now.

So I created a websocket service which sets up the Observable:

export class WebsocketService {
constructor() { }

private subject: Rx.Subject<MessageEvent>;

public connect(url): Rx.Subject<MessageEvent> {
   if (!this.subject) {
      this.subject = this.create(url);
   }
   return this.subject;
}

private create(url): Rx.Subject<MessageEvent> {
  let ws = new WebSocket(url);

  let observable = Rx.Observable.create(
     (obs: Rx.Observer<MessageEvent>) => {
        ws.onmessage = obs.next.bind(obs);
        ws.onerror = obs.error.bind(obs);
        ws.onclose = obs.complete.bind(obs);
        return ws.close.bind(ws);
     })
  let observer = {
     next: (data: Object) => {
        if (ws.readyState === WebSocket.OPEN) {
           ws.send(JSON.stringify(data));
        }
     }
  }
  return Rx.Subject.create(observer, observable);
 }

}

And then the RaceDataService which uses the WebSocketService to map the data to something useful:

export class RaceDataService {

   public dataStream: Subject<any>;

   constructor(wsService: WebsocketService) {
      this.dataStream = <Subject<any>>wsService
         .connect(ETS_URL)
         .map((response:any): any => {
            let data = JSON.parse(response.data);
            return data
         })
         .retry(3)
   }

   sendAction(action:WebsocketAction) {
      this.dataStream.next(action);
   }
}

Finally I subscribe to the observable in my race-display component like this:

constructor(private raceDataService:RaceDataService) {
    raceDataService.dataStream.subscribe(
      msg => {
        console.log("Response from websocket: ");
        console.log(msg);
        if(msg.RaceData) {
          this.HandleCurrentRaceData(msg.RaceData);
        }
      },
      err => {
        this.isConnected = false;
      },
      () => {
        this.sessionEnded = true;
      }
    )
  }

Now from what I've read, putting the .retry(3) after the setting up the observable object should ensure the connection is retried 3 times after failing. Eventually this will not be enough and I'll have to use retryWhen() to do some more advanced stuff, but at this point this simple retry isn't even working, the websocket connection is just attempted one time and thats it.

I'd be really grateful for any pointers.

From what I see that SignalR also requires server-side implementation. I'm using an ESP32 (Arduino-ish device) so I have to work with json data over websocket — Alex, Oct 20 '17 at 22:58
Where are you subscribing to dataStream? Using `map()` on the Subject just adds another step to the pipe, but I don't think it activates it - `subscribe()` will do that. — Richard Matsen, Oct 21 '17 at 00:08
This might help [SO: WebSocket: How to automatically reconnect after it dies](https://stackoverflow.com/questions/22431751/websocket-how-to-automatically-reconnect-after-it-dies) — Richard Matsen, Oct 21 '17 at 00:26
can you try comment out this line ws.onclose = obs.complete.bind(obs); and see if it works — Fan Cheung, Oct 21 '17 at 11:25
@FanCheung thanks for the suggestion but the result is the same, just 1 connection attempt — Alex, Oct 21 '17 at 17:24
@RichardMatsen Sorry I forgot to mention that, this is happening in the component which uses the race-data server. I've extended my question — Alex, Oct 21 '17 at 17:31

Richard Matsen · Answer 1 · 2017-10-21T19:34:37.230

Notice in the SO question WebSocket: How to automatically reconnect after it dies they talk about reconnecting when the socket closes.

In your code, retry(3) is only going to respond to events that occur via the line ws.onerror = obs.error.bind(obs), but those will only be errors that occur while the socket is live.

To retry connection errors with the Rx.retry(), you might try to hook websocket onClose to observable onError (ws.onclose = obs.error.bind(obs);), but Rx.retry() will just re-run wsService.connect(ETS_URL), which will just return the Subject, which (since it is now instantiated) will not call wsService.create() to perform the re-open.

I think you have to deal with the retry at the websocket level, not at the observable level - using functions attached to the onclose, as per the referenced question:

  ws.onclose = function(e) {
    console.log('Socket is closed. Reconnect will be attempted in 1 second.', e.reason);
    setTimeout(function() {
      connect();
    }, 1000);
  };

So, you have three types of error:

1) re-open prematurely closed connection

2) retry a failing initial open, for which you need try..catch around let ws = new WebSocket(url).

3) errors that occur during connection, emitted by ws.onerror. I'm not sure what they are, so not sure if the current strategy (passing them on to the observable) is best. Note that errors on the observable stream will close the stream, so if ws errors are ignorable (i.e a bad message occurs but subsequent messages are ok), then you don't want the stream to close.

How to get rxjs retry to work with websocket connection

1 Answers1