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?

Answer 1

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).

Answer 2

I prefer to use another operators, like concatMap() or switchMap() (difference between them is another topic). Especialy if you have store(a) operation in some db (SQL for example).

doOn[Next/Error/Completed] - is asynchronous callbacks, they best use case is some actions, like show/hide on UI progressbar.