Simplifying WebSockets in RxJS.
Using Redux-Observable for complex pipelines.
I’m an avid user of Redux-Observable, and while it masks a lot of the difficulties in using RxJS, you still need a deeper understanding of RxJS to handle really complex use cases. Using WebSockets, I’ll be showing you how you can make your own complex pipelines and understand them too!
WebSocket connection with RxJS in Node.js
In Node.js, setting up webSocket connection with RxJS is a bit difficult because unlike the browser, Node.js doesn’t have a native WebSocket object. With the ws library, all you need to do is override the WebSocketCtor value, and you’re good to go:
Pretty simple right? But this version doesn’t scale. All this does is create a webSocketConnection$ observable. It sets up only one connection, and you’d have to write the reconnection logic in your consumer.
Sometimes servers go down, and you want to make sure you can get back up ‘n running by connecting to the next available server.
Also keep in mind, webSocketConnection$ is “hot” meaning anything else that listens to it won’t start a new one but will listen in on this particular instance. Think of it like a singleton.
Knowing those restrictions, you might instead write your code like this:
We’ve increased the complexity level by utilizing recursive calls. Because we’re using timer (setTimeout under-the-hood), we don’t have to worry about throwing a stack overflow error.
What if you wanted even more functionality? What about cancellation? Say your service got what it needed and no longer needs to listen to this WebSocket connection? What if your callback needs to know when the connection dies and reconnects?
I’m constantly using more-complex functionality like this in frontend and backend applications. For example, when you leave view that cared about a WebSocket connection, you usually want to close the connection.
How can you do that with such a simple example?
For that, you’d need to listen to an event, or even better yet, use Redux-Observable.
Redux-Observable on the front- or backend
I’ve used Redux-Observable as both a frontend and backend framework for over 2 years now on just about every project that’s come my way. I even wrote my own Redux-Observable backend library for Node.js which I’ve used on many production projects such as Smart Home Services. The library allows you to use Redux and Redux-Observable without React in Node.js to create an event-sourced applications like Kafka, RabbitMQ, etc.
One of the packages I created makes it easy to spin up advanced WebSocket connections. Setting up WebSockets in Redux-Observable requires some thought so let’s go through it step-by-step.
Redux-Observable style
With Redux, you first call an action to make a WebSocket connection:
This is Redux-Observable though. We don’t just call actions in the middle of nowhere. For this example, we’ll start it when the app starts:
That’s better. When the APP_STARTED message comes through our pipeline, then we’ll start the connection.
This is a simpler version
As a disclaimer, everything you’re gonna see in this article is based on a single WebSocket connection. The code I’m using in production supports as many connections as you give it.
I was going to use that code in these examples, but it adds a significant amount of complexity and requires a deeper understanding of RxJS and Redux-Observable.
If you’re interested in the multi-socket code, checkout the code in this section my Redux-Observable-Backend repo.
Setting up the listener
Now that we have an action to tell us we need to make a WebSocket connection, now we need to write the code to actually do it. Don’t worry about understanding it all now; we’ll go through it in pieces:
This should initially look complex, but after breaking it down in parts, it’ll become much easier to grok.
Breaking it apart
Starting simple
Without this code, making a WebSocket connection is pretty useless since this is the entire reason the rest of the logic exists.
When a webSocketConnection$ message comes in, call the receivedWebSocketMessage action creator to format and make that message available to all of Redux and Redux-Observable.
The most-clever RxJS operator: `startWith`
The one most-impressive operators in RxJS is startWith. I’ve used this operator in so many crazy ways. This epic actually uses it twice!
Once we’ve created a webSocketConnection, we need to send out a message to listeners saying the connection’s ready to go. Using startWith, we can immediately push the CONNECTION_READY action through the pipeline to be automatically dispatched as soon as the observable gets subscribed; even before any data comes through.
In the other place we’re using startWith, it’s acting as the equivalent of:
const someFunction = () => { /* do stuff */ }someFunction()
This part code the code says “once this observable is subscribed, immediately start at this point in time and begin execution”. This lets us reuse the reconnection listener create the webSocket connection on subscribe. It’s pretty ingenious and exactly why I think startWith is so powerful.
Reconnect on connection error
This part subscribes to the webSocketConnection$ observable and catches any connection errors just like we did in our Node.js example. It even setups up a timer too!
In no way are we recursively calling a function; instead, we’re dispatching an action RECONNECT_TO_SERVER which just so happens to be the action we’re listening to on this inner observable.
Cancellation
One of the most-important reasons to use RxJS is cancellation. Unlike promises and callbacks, which have no built-in methods for cancellation, there are many ways to cancel observables.
When an observable gets canceled in RxJS, it could either be through an error or completing the observable. In our case, we’re using takeUntil to listen for an action that ends up completing the observable at various stages of execution.
Listening to action$, we can check for various action types and keep our observables running until an action type comes in that complete it such as DISCONNECT_FROM_SERVER.
With takeUntil on RECONNECT_TO_SERVER, we’re saying “complete this connection observable because we’re reconnecting”.
The very first instance of takeUntil is only listening for DISCONNECT_FROM_SERVER. This is because it’s the listener for RECONNECT_TO_SERVER. If we had it cancel after RECONNECT_TO_SERVER, it’d cancel as soon it started and would never start the WebSocket connection.
Why so many cancellations?
There are three places we’re listening for actions that have the RECONNECT_TO_SERVER and DISCONNECT_FROM_SERVER types. While you’d think that when the parent observable completes, so do the child observables, but that’s not the case.
The observables subscribed from catchError and switchMap don’t complete until they’re done processing. If the parent sends a value that triggers the switchMap operator and the parent completes, the observable in switchMap is still active until it also completes. That’s why we need to define takeUntil operators in those three places.
See it all again
To recap, we first looked at creating WebSocket observables with RxJS in Node.js then went into how you’d use Redux-Observable to do the same either in the browser or Node.js.
Then we broke that epic in pieces to learn how WebSocket messages are converted to actions, why startWith is super useful, how catchError helps us handle reconnections, and how cancellation is possible but requires some thought.
Conclusion
Looking through the code a second time, does it make more sense now? Is it clear why it’s written this way and the benefits we gain? Feel free to comment your thoughts after going through all this. I’d love to hear what you learned and how this could help you in your own project.
If you’d like to see a real-world in-production-now example of this same code that allows for multiple WebSocket connections, look no further than this section my Redux-Observable-Backend repo.
More Reads
If you liked what you read, please checkout my other articles on similar eye-opening topics:
