const ws = Observable.webSocket('ws://…');
ws.subscribe(
  message => console.log(message),
  error => console.log(error),
  () => {},
);

I want to observe my WebSocket connection with RxJS. Reacting to onmessage events by subscribing to the observable works like a charm. But how can I access the onopen event of the WebSocket? And is it possible to trigger the WebSocket .close() method? RxJS is pretty new to me and I did research for hours, but maybe I just don't know the right terms. Thanks in advance.

const ws = Observable.webSocket('ws://…');
ws.subscribe(
  message => console.log(message),
  error => console.log(error),
  () => {},
);

I want to observe my WebSocket connection with RxJS. Reacting to onmessage events by subscribing to the observable works like a charm. But how can I access the onopen event of the WebSocket? And is it possible to trigger the WebSocket .close() method? RxJS is pretty new to me and I did research for hours, but maybe I just don't know the right terms. Thanks in advance.

• javascript
• websocket
• rxjs
• rxjs5

2 Answers 2

Looking at the sourcecode of the Websocket there is a config object WebSocketSubjectConfig which contains observables which you can link to different events. You should be able to pass a NextObserver typed object to config value openObserver like so:

const openEvents = new Subject<Event>();
const ws = Observable.webSocket({
  url: 'ws://…', 
  openObserver: openEvents 
});

openEvents
  .do(evt => console.log('got open event: ' + evt))
  .merge(ws.do(msg => console.log('got message: ' + msg))
  .subscribe();

The link of @Mark van Straten to the file is dead. The updated link is here. I also wanted to highlight the usage as suggested in the docs. A in my opinion better copy and paste solution to play around with:

import { webSocket } from "rxjs/webSocket";

webSocket({
  url: "wss://echo.websocket",
  openObserver: {
    next: () => {
      console.log("connection ok");
    },
  },
  closeObserver: {
    next(closeEvent) {
      // ...
    }
  },
}).subscribe();