Integrating redux-observable and RxJS into a small part of a react/redux app


In this example I will demonstrate how to hook up redux-observable for a single action, into an existing react/redux app which currently uses redux-thunk throughout.

The problem

We recently encountered a problem whereby we needed to wait for a response from the server on page one before submitting another request on page two. This doesn't sound like much of an issue, but the problem was the request was slow, and the second request, (a POST to update a record), did not have the necessary data needed from the first request. The UX would also suffer if we needed to show a loading state on either of the first or second pages.


‘Comment system’, whereby you leave a written comment on an item on the first page, and a star rating on the second page.

Page one > Add a written comment > Click next button > POST to server > Server responds with new comment Id.

Page two > Add a star rating (1–5) > Click next button > POST to server (with the ID from the initial post so the server knows which record to update) > Server responds with comment Id

The request on page two will fail given that the request has no comment Id in the payload.

Why RxJS

With RxJS requests can be treated as streams of data. This means we are able to not only cancel existing requests if required, but also to control in what order new requests are dealt with. In our example we want to use RxJS to concat requests, so that only when the initial request on page one has responded, do we fire off the second request on page two.

Why redux-observable

redux-observable is redux middleware, (in the same way redux-thunk is), that allows RxJS related code to be run when an action is fired.


npm i -S rxjs redux-observable

Async function to save comments

We will create a fake service call which will return a promise. It will delay the response by 2 seconds and send back an Id of 1 to simulate a successful save of a comment.

const saveAsync = request =>
new Promise(resolve => {
console.log('request data:', request);
setTimeout(() => {
resolve({id: 1, request});
}, 2000);

Actions and reducers for saving comments

We will have an action to save a comments (type: SAVE_COMMENT), and then an action which gets fired when the comment is saved on the server, (COMMENT_SAVED). When the comment is saved the reducer will update the comment, (Id), in state. We also add a hitNum property which is just a random number and will confirm the requests are getting fired in the correct order later on.

saveComment = comment => ({
hitNum: Math.floor(Math.random() * 1000) + 1,
const commentSaved = (state, action) =>
Object.assign({}, state, { action.comment });
['COMMENT_SAVED']: commentSaved

Create epic middleware

This is a function that takes, and returns, a stream of actions. Feed it to the redux-observable createEpicMiddleware function. Add the middleware when creating your redux store.

import { createStore, applyMiddleware } from 'redux';
import { createEpicMiddleware } from 'redux-observable';
import 'rxjs'; //import globally, (not advised, demo only)
const commentEpic = (action$, state) => {};
const epicMiddleware = createEpicMiddleware(commentEpic);
//Add this middleware to your redux create store
const store = createStore(

Expanding the epic

We need to tell the comments epic to act on the stream of SAVE_COMMENT actions and return COMMENT_SAVED actions. Variables appended with a dollar depict an observable. We concatMap over each save comment action. This is what allows for the response of the first request to return before the second request is fired. So for each save comment action, we get the comment Id out of state, (if there is one), and assign it to the actions payload. We then call the ‘saveCommentObservable’, (which we will be writing next), passing in the comment payload. This will return a deferred observable which we will map over so we can return the ‘COMMENT_SAVED’ action.

const commentEpic = (action$, state) =>
.concatMap(action => {
const id = get('', state.getState());
const payload = Object.assign({}, action.comment, { id });
return saveCommentObservable(payload)
.map(comment => ({
hitNum: action.hitNum,

Creating the save comment observable

This is a function that calls the saveAsync function which is wrapped in two observables. We start by returning a defer observable which will wrap a newly created observable. This is needed because we only want to create the inner observable when the observer subscribes, (lazy). Within the inner created observable we call the saveAsync action and only call complete when the promise resolves.

const saveCommentObservable = payload =>
Rx.Observable.defer(() =>
Rx.Observable.create(obs => {
let cancelled = false;
.then(comment => {
return () => {
cancelled = true;


Now running the integration and submitting the two pages. You will see that the saveAsync function has logged correctly:

request data: Object {rating: 1, id: undefined}
request data: Object {rating: 1, id: 1}

And the actions have fired in the correct order, (note, you will not see the Id in the second action as the action does not pass the Id, it is got from state in the comment epic):


Software dev — Javascript, node, express, mongo, react, redux, rxjs, es6, ramda