Buffered alternate parallel and non-parallel executions about rxjavaextensions HOT 10 CLOSED

Why don't you use the available operators to switch between modes?

datasource
.to(ParallelFlowable::from)
.runOn(Schedulers.computation())
.doOnNext(v -> { })
.sequential()
.observeOn(Schedulers.single())
.buffer(1000)
.doOnNext(list -> { })
.subscribe();

from rxjavaextensions.

Wonderful !
Very impressive power !
I forgot to say that the first computation is IO intensive. It's better to use another type of scheduler ?

(I'm still studying rxjava.)

from rxjavaextensions.

As you wish, the parallelism level is determined by the from parameter and not the Scheduler:

.to(f -> ParallelFlowable.from(f, 3))
.runOn(Schedulers.io())

from rxjavaextensions.

... and in this way the first doOnNext lambda is not executed when the second one is ?

from rxjavaextensions.

Stages will run in parallel, why do you have that requirement? RxJava is about executing in a flow and not blocking out things; backpressure is there to limit memory usage if that's the problem.

from rxjavaextensions.

I need to alternate the two stages.
Stage one is read only
Stage two is when results are saved

I need to isolate them, the application is much more simple and robust this way.
What I need is something like a blocking-output-buffer for the doOnNext operator. When it's full it blocks the doOnNext until the buffer is consumed

from rxjavaextensions.

The individual items should be isolated from each other anyway.

What I need is something like a blocking-output-buffer for the doOnNext operator. When it's full it blocks the doOnNext until the buffer is consumed

This makes no sense in reactive.

Here is an alternative that works on a chunk of 1000 elements:

datasource
.window(1000)
.flatMap(w ->
    ParallelFlowable.from(w)
    .runOn(Schedulers.io())
    .doOnNext(item -> { })
    .sequential()
    .toList()
    .doOnNext(list -> { })
    , 1)
.subscribe();

Here processing the items parallel happens before collecting them into a list and once that happens, you can process the list and there won't be any parallel execution from the first doOnNext nor there won't be another window started before you have finished with the list.

from rxjavaextensions.

OK that is a solution.
So it's not possible without buffering at start.
What's wrong with an operator that in loop do the following ?
do something with item,
collect item
emit items when the buffer is full

from rxjavaextensions.

What's wrong with an operator that in loop do the following ?

That's the imperative thinking about processing data. In reactive, there is no explicit loop but flow of data, aggregation, element processing, etc.

from rxjavaextensions.

ok... I'm still thinking in imperative mode.

from rxjavaextensions.