Branching observables and concurrency configuration in Rx

Question

Consider this example:

observable
  .Where(somePredicate)
  .ObserveOn(TaskPoolScheduler.Default)
  .SubscribeOn(TaskPoolScheduler.Default)
  .Subscribe(x => { });

Is the Where() executed in the task pool, or only the subscribe func?

Following on:

var newObservable = observable.
  .Where(somePredicate)
  .ObserveOn(TaskPoolScheduler.Default)
  .SubscribeOn(TaskPoolScheduler.Default);

newObservable.Subscribe(x => { });
newObservable.Subscribe(x => { });

In this example, do both subscriptions retain the concurrency settings of newObservable? In that sense, can the creator of an observable define a different concurrency default (vs the Rx default) and pass it on to users of the observable, unless overriden?

Edit:

Now, if:

newObservable
  .Where(somePredicate2)
  .Subscribe(x => { })

Is this now executed on the task pool too?

As for #1, everything gets executed on the threadpool. Your application would block while waiting for events if `Where` executed on the main thread. — Panagiotis Kanavos, Apr 20 '15 at 10:22
There's a great introduction to concurrency in Rx [here](http://www.introtorx.com/Content/v1.0.10621.0/15_SchedulingAndThreading.html#SchedulingAndThreading) — Roland Pheasant, Apr 20 '15 at 10:30
As for #2 check [ObserveOn and SubscribeOn - where the work is being done](http://stackoverflow.com/questions/20451939/observeon-and-subscribeon-where-the-work-is-being-done). — Panagiotis Kanavos, Apr 20 '15 at 10:47
@PanagiotisKanavos so if I understood it correctly, the `ObserveOn` and `SubscribeOn` apply to the part of the stream after the calls. Correct? If you can write a short summary (the other thread is great for an in-depth understanding) as an answer, I can accept it. — georgiosd, Apr 20 '15 at 11:09

score 1 · Answer 1 · edited May 23 '17 at 12:22

As per the comments, see my fuller answer ObserveOn and SubscribeOn - where the work is being done for details, but:

Fundamentally, you need to understand that ObserveOn and SubscribeOn are decorators that only affect the downstream behaviour of an observable (the callers of Subscribe and receivers of OnXXX) - they don't magically change the internal behaviour of the observable they wrap.

Where the Where executes is mostly dependent on the implementation of observable, which you don't share. Certainly, ObserveOn has no impact and SubscribeOn may have an impact if and only if the observable implementation yields events on the subscriber's thread - but don't use it to try and control this, that's not what it is for.
In the follow on, I think you are asking if subscribers to newObservable have the behaviour of ObserveOn and SubscribeOn? In which case, yes. Again I suspect taking the time to read the above answer will make this very obvious.
When you add the Where on the newObserable, the thread on which the subscriber is called won't be affected - and it's purely the implementation of Where that determines this, nothing to do with any upstream ObserveOn or SubscribeOn.

Bottom line, there is a summary in the top of my answer - that's the best TL;DR I have for you.

I'm still rather confused. I'll read the longer answer again when my head is not spinning and retort :) — georgiosd, Apr 20 '15 at 15:16

Branching observables and concurrency configuration in Rx

1 Answers1