RxJs is one of my favorite Javascript libraries. It's almost an update to the language itself. In a recent role, we built an awesome data pipeline that handled sorting, filtering, aggregating and paginating for multiple large datasets in the user's device using RxJs. We also used a Redux store to maintain the application's state and IndexedDB to store the large datasets in the user's device. With this and a few other tricks we were able to deliver a progressive web application that worked offline.
In this posts, I'd like to share with you a custom Redux middleware function that we developed to easily handle Observables within Redux actions. There are a few libraries in NPM that advertise the ability to do this but we found that the solution was simple enough that a security audit of a third-party library was just a waste of time.
We wanted to keep our React view layer pure by keeping all business logic out of it. The application rendered the state of the Redux store which included the query necessary to extract the visible records from IndexedDB. Any actions from the user where mapped to a dispatch to the store which could be a simple action or an observable action. We skipped accepting functions as a thunk action since this is basically the same thing as an observable action but the middleware actually ignores these as well so it's safe to combine with Redux-Thunk.
We determined that there existed two types of logic that we wanted to store in observables.
- Business
- Asynchronous
That's really what RxJs is all about. Rather than have complex or several middlewares, intermediate actions and complex reducers to handle network requests and other business tasks, we delegated all of this to RxJs which allows us to isolate logic and compose it seamlessly. The Redux store focuses on mapping actions to state and notifying the React application of changes in state. With RxJs we can map a user dispatched action to a write to the Redux store all within a single observable interface which seamlessly hides away the asynchronous scheduling and the several intermediate tasks.
Ok, so enough talk. Here's a working example:
The above is a simple script that uses a Redux store to add numbers together and return the total sum. It also keeps track of errors and has a loading flag so that the user wont receive intermediate values.
If you take a look at the counterReducer, we have four action types to keep in mind. The _RESET action type is emitted synchronously so the reducer will receive this notification before any of the notifications from the observable. This can be used to reset some state such as clearing the errors, clearing the counter, and setting the loading flat to true.
The observerMiddleware will handle observable actions.
By default, an Observable will attempt to execute synchronously and switch to asynchronous processing once it encounters a Promise or any other type of asynchronous operation. This might lead to the _NEXT, _ERROR and _COMPLETE notifications being emitted before the _RESET which might cause the _RESET notification to clear out the state after we've updated it. We need to change the default scheduler on the Observable to prevent this. Luckily for us, RxJs provides a very simple way to guarantee that an observable is processed asynchronously. All we have to do is apply the observeOn operator with the asapScheduler to the observable and presto!
Now, our observable will begin processing after the _RESET notification and will emit every value as a _NEXT notification. Our reducer will update the state of the counter for each of these notifications.
Finally, if the observable emits an _ERROR notification, our reducer will update the state with the error. Otherwise, the observable will emit a _COMPLETE notification which our reducer will process to set the loading flag to false. Now our user-facing application can remove the loading indicator and display the total sum (or the error message).
We'll test this out by creating an action generator named addNumbers that receives any amount of numbers as arguments and dispatches an action with an observable payload that emits these numbers in sequence before completing. Given the numbers 1 through 4, we expect a sum of 10. If we run the above script, we can see that once the loading flat is set to false and the value of counter will be set informing us that the process has finished loading and the total sum is 10.
Top comments (0)