Rx Schedulers

[benjchristensen]

Implementation of Rx Schedulers (http://msdn.microsoft.com/en-us/library/hh242963(v=vs.103).aspx) ... probably to go into the rx.concurrency package (https://github.com/Netflix/RxJava/tree/master/rxjava-core/src/main/java/rx/concurrency).

[abersnaze]

Interesting quote from that page
"If you do not use the overload which takes a scheduler as an argument, Rx will pick a default scheduler by using the principle of least concurrency. This means that the scheduler which introduces the least amount of concurrency that satisfies the needs of the operator is chosen. For example, for operators returning an observable with a finite and small number of messages, Rx calls Immediate. For operators returning a potentially large or infinite number of messages, CurrentThread is called. For operators which use timers, ThreadPool is used."

[benjchristensen]

How would an operator know how many messages it is going to have? Only the origin Observable could know that - a given operator along the chain won't know so how does this get accomplished?

[benjchristensen]

I guess this would mean that things like "toObservable(1, 2, 3, 4)" are a known thing and can be done immediately without a thread, but merging 4 unknown sequences can't be known.

Obviously when Timers are used a Thread of some kind is needed (java.util.Timer or another implementation like this: https://github.com/Netflix/Hystrix/blob/master/hystrix-core/src/main/java/com/netflix/hystrix/util/HystrixTimer.java).

For the majority of cases though where the "cost" of the observable sequence is unknown I don't know that I like automatically spawning them off on threads. It has worked well for the Netflix API to leave that choice to the origin of the observable (to be synchronous or asynchronous, one thread or many, etc).

The introduction of Schedulers makes perfect sense when an app is dealing with data structures, their own synchronous IO operations or CPU bound computations, but it becomes awkward when consuming from a 3rd party API exposing Observables who can and will make their own choice about being synchronous or asynchronous.

In fact, that's been a huge part of the draw to Rx is that the API doesn't need to change when the backend implementations moves between synchronous and asynchronous for whatever reason it may choose to do so.

If the Observable is already asynchronous it would be inefficient to spin up another thread that in turn blocks on an async call.

Other than documentation on the API calls that return Observables is there a better way to handle that scenario?

I can imagine a scenario where some apps (such as the Netflix API) may want to disable any use of Schedulers so the origin retains control since Rx has allowed us to decouple the writing of business logic from the decisions of concurrency.

Before flaming me ... I DO like schedulers, it's very powerful and we will definitely get them added, I just have some questions about balancing that power (and inevitable inefficiencies of making poor scheduler choices) with the elegant simplicity of Rx Observables without them where concurrency is not a thought - everything is just asynchronous.

I'm interested in all of your perspectives so please chime in.

[abersnaze]

The docs say "least amount of concurrency" which I interpreted to mean that if the amount of work is unknown that it would it default to immediate. We could still manipulate the defaults and probably ignore Schedulers passed in through a strategy.

[benjchristensen]

Yes, I think a strategy pattern will be needed to accomplish the Netflix API use case.

[benjchristensen]

Some thoughts while working on the design of this:

• The Scheduler interface should be capable of supporting different sources of concurrency such as Executors, Threads, Actors, EventLoops
• We should be capable of supporting rxjava-contrib modules with new types of Schedulers such as for Akka/Scala Actors
• We need the ability (via plugins probably) of a system to override or prevent usage of Schedulers where they want. For example, if a system doesn't want client code starting threads they should be able to intercept and ignore or throw UnsupportedOperationException.

[benjchristensen]

Sections 6.9 through 6.12 of the Rx Design Guidelines PDF (http://go.microsoft.com/fwlink/?LinkID=205219) should be read by anyone involved in Schedulers design and implementation.

[jmhofer]

There's also this video explaining the motivation behind introducing schedulers in Rx.

[benjchristensen]

Good video ... thanks for the link.

[benjchristensen]

First round of Schedulers implementation committed via pull request #225 contribued by @mairbek.

It implements ObserveOn (#11) and SubscribeOn (#12).