ReactiveX Redux

Everyone who works with Redux sooner or later faces the problem of asynchronous actions. But a modern application cannot be developed without them. These are http-requests to the backend, and all sorts of timers / delays. The creators of Redux themselves say unequivocally - by default, only synchronous data-flow is supported, all asynchronous actions must be placed in middleware.

Of course, this is too verbose and inconvenient, so it’s hard to find a developer who uses only native native middleware. Libraries and frameworks such as Thunk, Saga and the like always come to the rescue.

For most tasks, they are quite enough. But what if a bit more complicated logic is needed than to send one request or make one timer? Here is a small example:
')

async dispatch => { setTimeout(() => { try { await Promise .all([fetchOne, fetchTwo]) .then(([respOne, respTwo]) => { dispatch({ type: 'SUCCESS', respOne, respTwo }); }); } catch (error) { dispatch({ type: 'FAILED', error }); } }, 2000); } 

It is painful to even look at such code, and it is simply impossible to maintain and expand. What to do when you need more complex error handling? What if you need to repeat the request? And if I want to reuse this function?

My name is Dmitry Samokhvalov, and in this post I will tell you what the Observable concept is and how to put it into practice in conjunction with Redux, and I will compare all this with the capabilities of Redux-Saga.

As a rule, in such cases take redux-saga. OK, rewrite the saga:

 try { yield call(delay, 2000); const [respOne, respTwo] = yield [ call(fetchOne), call(fetchTwo) ]; yield put({ type: 'SUCCESS', respOne, respTwo }); } catch (error) { yield put({ type: 'FAILED', error }); } 

It became noticeably better - the code is almost linear, it looks better and is read. But it is still difficult to expand and reuse, because the saga is just as imperative as thunk.

There is another approach. This is exactly the approach, and not just another library for writing asynchronous code. It is called Rx (they are also Observables, Reactive Streams, etc.). Let's use it and rewrite the example on Observable:

 action$ .delay(2000) .switchMap(() => Observable.merge(fetchOne, fetchTwo) .map(([respOne, respTwo]) => ({ type: 'SUCCESS', respOne, respTwo })) .catch(error => ({ type: 'FAILED', error })) 

The code did not just become flat and decreased in volume; the very principle of describing asynchronous actions changed. Now we do not work directly with requests, but perform operations on special objects called Observable.

Observable is conveniently presented as a function that gives a stream (sequence) of values. Observable has three main states - next (“give the next value”), error (“an error has occurred”) and complete (“the values ​​have ended, there is nothing more to give”). In this regard, it is a bit like Promise, but it differs in that it is possible to iterate over these values ​​(and this is one of the Observable superpowers). You can wrap anything in Observable — timeouts, http requests, DOM events, just js objects.



The second superpower is Observable operators. An operator is a function that accepts and returns Observable, but performs some operations on a stream of values. The closest analogy is map and filter from javascript (by the way, there are such operators in Rx).



The most useful for me personally were the zip, forkJoin and flatMap operators. On their example it is easiest to explain the work of the operators.

The zip operator works very simply - it accepts several Observables as input (no more than 9) and returns in the form of an array the values ​​that they emit.

 const first = fromEvent("mousedown"); const second = fromEvent("mouseup"); zip(first, second) .subscribe(e => console.log(`${e[0].x} ${e[1].x}`)); //output [119,120] [120,233] … 

In general, zip work can be represented by the scheme:



Zip is used if you have several Observables and you need to consistently receive values ​​from them (if they can be emitted at different intervals, synchronously or not). It is very useful when working with DOM events.

The forkJoin operator is similar to zip with one exception - it returns only the latest values ​​from each Observable.



Accordingly, it is reasonable to use it when only final values ​​from the stream are needed.
A bit more complicated is the flatMap operator. It takes an Observable as input and returns a new Observable, and maps values ​​from it to the new Observable, using either the selector function or another Observable. It sounds confusing, but the diagram is pretty simple:



More clearly in the code:

 const observable = of("Hello"); const promise = value => new Promise(resolve => resolve(`${value} World`); observable .flatMap(value => promise(value)) .subscribe(result => console.log(result)); //output "Hello World" 

The most common flatMap is used in backend requests, along with switchMap and concatMap.
How can Rx be used in Redux? For this there is a remarkable library redux-observable. Its architecture looks like this:



All Observable, operators and actions on them are made out in the form of special middleware, which is called epic. Each epic takes an action as input, wraps it in an Observable and must return an action, also in the form of an Observable. It is impossible to return an ordinary action, it creates an infinite loop. Let's write a small epic, which makes a request to api.

 const fetchEpic = action$ => action$ .ofType('FETCH_INFO') .map(() => ({ type: 'FETCH_START' })) .flatMap(() => Observable .from(apiRequest) .map(data => ({ type: 'FETCH_SUCCESS', data })) .catch(error => ({ type: 'FETCH_ERROR', error })) ) 

It is impossible to do without comparing redux-observable and redux-saga. It seems to many that they are close in functionality and capabilities, but this is not at all the case. Sagas are a completely imperative tool, essentially a set of methods for working with side effects. Observable is a fundamentally different style of writing asynchronous code, if you like, another philosophy.

I wrote a few examples to illustrate the possibilities and approach to solving problems.

Suppose we need to implement a timer that will stop in action. Here is how it looks on the sagas:

 while(true) { const timer = yield race({ stopped: take('STOP'), tick: call(wait, 1000) }) if (!timer.stopped) { yield put(actions.tick()) } else { break } } 

Now use Rx:

 interval(1000) .takeUntil(action$.ofType('STOP')) 

Suppose there is a task to implement a request with cancellation on sagas:

 function* fetchSaga() { yield call(fetchUser); } while (yield take('FETCH')) { const fetchSaga = yield fork(fetchSaga); yield take('FETCH_CANCEL'); yield cancel(fetchSaga); } 

Rx makes things easier:

 switchMap(() => fetchUser()) .takeUntil(action$.ofType('FETCH_CANCEL')) 

Finally, my favorite. Implement the request to api, in case of failure, make no more than 5 repeated requests with a delay of 2 seconds. Here is what we have on the sagas:

 for (let i = 0; i < 5; i++) { try { const apiResponse = yield call(apiRequest); return apiResponse; } catch (err) { if(i < 4) { yield delay(2000); } } } throw new Error(); } 

What happens on Rx:

 .retryWhen(errors => errors .delay(1000) .take(5)) 

If we sum up the pros and cons of the saga, we get the following picture:



Sagas are easy to learn and very popular, so you can find recipes in the community for almost all occasions. Unfortunately, the imperative style makes it difficult to use the saga truly flexible.

Rx has a completely different situation:



It may seem that the Rx is a magic hammer and a silver bullet. Unfortunately, this is not the case. The threshold for entering Rx is noticeably higher, so it is harder to introduce a new person into a project that actively uses Rx.

In addition, when working with Observable, it is especially important to be attentive and always understand well what is happening. Otherwise, you may stumble upon non-obvious errors or undefined behavior.

 action$ .ofType('DELETE') .switchMap(() => Observable .fromPromise(deleteRequest) .map(() => ({ type: 'DELETE_SUCCESS'}))) 

Once I wrote an epic that did a fairly simple job — for each action with a type of 'DELETE', an API method was called that would delete the element. However, testing has problems. The tester complained about the strange behavior - sometimes when you clicked the delete button, nothing happened. It turned out that the switchMap operator supports the execution of only one Observable at a time, a kind of protection against race condition.

As a result, I will cite several recommendations that I follow myself and urge you to follow everyone who starts working with Rx:

• Be careful.
• Study the documentation.
• Check in the sandbox.
• Write tests.
• Do not shoot a cannon on the sparrows.