How should I handle side-effects in RxJava?

Question

I've got an Observable which emits a certain object S. This object internally has a list of objects, each containing a set of values. I need this data flat (unwinding all the inner collections into a long sequence of flat objects A), store it in a database if it doesn't already exist, and then reduce the sequence of objects A back to an object T with a similar structure to starting object S, which I need to pass along.

I figured that altering state outside of your function is not a good idea, so writing this as one big Observable transformation like this is a no-go (especially since there is also a blocking database call within the second map):

sObservable
    .map(turnSIntoAFn)     // <-- Actually more complex
    .map(a -> {
         store(a);
         return a;
    })
    .map(turnAIntoTFn)     // <-- Actually more complex
    .subscribe(...);

I then figured that I should limit my side-effects to the subscriber. That would leave me with either one of the following situations:

Transforming Observable to Observable, then subscribing with a subscriber that throws these A's into the database. Then query the database, get the A's in an Observable (hooray for MongoDB's Rx driver), transform them to T's and pass them along (with a subscriber)
Using the source Observable, and doing two things simultaneously:
1. like the previous step, transform it into an Observable and store the A's in the database.
2. transform Observable directly into Observable, passing it along optimistically.

For now, the first option looks like the more appealing one, although it needs more database actions than both the dirty version and the second of the alternatives. Is there not a better way of doing this sort of thing (doing something useful with data, then pass it along) without either incurring more strain on resources and altering state outside of my functions?

Is the return value of the second map in any way dependent on the external state, e. g. is it re-reading from the DB? Because if it isn't (and I think, it shouldn't) then you could just use doOnNext instead. This makes it clear that what's happening in there is just a side effect with no impact on what is happening below... — david.mihola
I am having the same question about side-effects. In this case, I may get the observable from obsevable.map(turnSIntoAFn) and use it for subscribeing to store a and createing the observable to map. I don't know if it would be a good idea. — otal

Dan Lew Dan Lew · Accepted Answer · 2016-02-05T15:30:06

You should consider using the do operators - in this case, doOnNext().

doOn[Next/Error/Completed] are like little side effecting subscriptions that you can put in the sequence. It's not part of the transformation (since they can't transform the data).

How should I handle side-effects in RxJava?

2 Answers