Create queue flows with Rxjava

Question

I must create a stream of execution, in which a series of Observables must be executed, some being in parallel and others awaiting the execution of the ones that are in parallel to be executed. These Observables manipulate data that is passed to the others. For example, Observable 2 handles data that must be consumed by Observable 4 later.

In summary, the flow should be something like:

Time ------------------------->

-Observable 1 (Get data on database. Parallel with 2)
-Observable 2 (manipulate existing ArrayList to future use)
-----Observable 3 (Get data on another database. After 1 AND 2 executions, because using data from both)
---------Observable 4 (manipulate result of 3, using list result to create another list. After 3)
-------------Observable 5 (with result of 4, i will insert log data on database. After 4 and Parallel with 6)
-------------Observable 6 (create a list of result to return on actual method)

Some observables in parallel and others waiting for the finalization to execute, and those that wait, can be one, or more of one, to be executed also in parallel.

I have been trying in various ways, creating lists of Observables and running in .zip (with .observeOn (Schedulers.computation ()) in each observable), I have also tried with .zipWith. But without success.

What are the datatypes involved? Are these Observables dependent on each other's results or do you perform manipulations only as side-effects? — akarnokd
Some observables are simple list manipulation processes, others get data from some datasource (database). And the data must be passed from one observable to another after treated or collected from the database. — touchmx
The answer depends on the datatypes, the number of items of the observables and how they have to take each other's elements. Please provide more information about each Observable. — akarnokd

akarnokd akarnokd · Accepted Answer · 2018-04-10T21:12:56

Let's assume you have these methods:

Single<List<DataType1>> getData1()

Single<List<DataType2>> manipulateList2(List<DataType2> list2)

Single<List<DataType3>> getData3(List<DataType1> list1, List<DataType2> list2)

Single<List<DataType4>> manipulateList4(List<DataType3> list3)

Completable insertLog(List<DataType4> list4)

Single<List<ResultType>> createResult(List<DataType4> list4)

You can then combine these as required

Single<List<ResultType>> result = Single.zip(
    getData1().subscribeOn(Schedulers.io()),
    manipulateList2(sourceList2).subscribeOn(Schedulers.computation()),
    (list1, list2) -> getData3(list1, list2)
)
.flatMap(v -> v)
.flatMap(list3 -> manipulateList4(list3))
.flatMap(list4 -> 
    insertLog(list4).subscribeOn(Schedulers.io())
    .andThen(createResult(list4).subscribeOn(Schedulers.computation()))
);

return result;

zip allows running multiple sources of the same base reactive type at once, flatMap acts as a continuation and andThen let's you run a Completable and a Single together.

Create queue flows with Rxjava

1 Answers