.net Observable 'ObserveOn' a background thread

Question

I am trying to implement a simple Observer pattern using .net Observable class. I have code that looks like this:

Observable.FromEventPattern<PropertyChangedEventArgs>(
    Instance.User,
    "PropertyChanged")
          .Where(e => e.EventArgs.PropertyName == "FirstName")
          .ObserveOn(Scheduler.ThreadPool)
          .Subscribe(search => OnFirstNameChanged(search.EventArgs));

Observable.FromEventPattern<PropertyChangedEventArgs>(
    Instance.User,
    "PropertyChanged")
          .Where(e => e.EventArgs.PropertyName == "LastName")
          .ObserveOn(Scheduler.ThreadPool)
          .Subscribe(search => OnLastNameChanged(search.EventArgs));

I want the observers to run on a background thread, but I want them to all run on the same background thread (for our real implementation, it will be too complicated to have every listener on a different thread).

i.e. I want all of the OnXXXChanged logic to be performed on a thread other than the UI thread, but instead of Observing on the entire threadpool, I want to make sure they run in the correct order, on the same thread.

How should the above be modified?

Also, on a somewhat related note, are there any good sample code examples using the Observable class to implement this pattern?

Answer 1

You should create an EventLoopScheduler and use that single instance in all calls to ObserverOn:

var scheduler = new EventLoopScheduler(ts => new Thread(ts));

... .ObserveOn(scheduler). ...

The thread created by the factory method is the thread used to schedule the execution on. By leaving the property ExitIfEmpty set to false this thread will not terminate even if there is nothing to do meaning that it will be reused for every call.

However, you could also consider using Scheduler.NewThread. Using that scheduler will allow the thread to terminate if there is nothing more to do. When more work is queued up by ObserverOn a new thread will be created but only a single thread should ever exists meaning that you don't have synchronize different observers.

The threads created by EventLoopScheduler (which is used by Scheduler.NewThread) are named Event Loop #. You will see these names in the debugger.

Answer 2

.ObserveOn(Scheduler.ThreadPool) takes a thread scheduler which dictates the thread that the observation runs on. It looks like for a single thread you want to use EventLoopScheduler, rather than ThreadPool.