Diving deeper in Reactive Extensions, concurrency

Oct 9, 2012

I’ve started using Reactive Extensions, Rx library without digging too much in it’s internals. Had a case where I wanted to filter a stream of events and decided I’d try something new for me, Rx. But as it always happens with powerful and flexible tools, starting with few superficial articles and examples from the Web without diving deeper in the concepts is a pretty bad idea.

This time I’ve been bitten by concurrency issues. And the best resource I’ve found so far is the Introduction to Rx ebook by Lee Campbell. It costs only around a buck on Amazon and it’s available for free from the site.

I’ve jumped straight to the Chapter 4, Concurrency and for surely will pass without a hurry over the rest of the book. This chapter has an nice check list about best threading options for some scenarios:

UI Applications

The final subscriber is normally the presentation layer and should control the scheduling.

Observe on the DispatcherScheduler to allow updating of ViewModels

Subscribe on a background thread to prevent the UI from becoming unresponsive

If the subscription will not block for more than 50ms then

Use the TaskPoolScheduler if available, or

Use the ThreadPoolScheduler

If any part of the subscription could block for longer than 50ms, then you should use the NewThreadScheduler.

Service layer

If your service is reading data from a queue of some sort, consider using a dedicated EventLoopScheduler. This way, you can preserve order of events

If processing an item is expensive (>50ms or requires I/O), then consider using a NewThreadScheduler

If you just need the scheduler for a timer, e.g. for Observable.Interval or Observable.Timer, then favor the TaskPool. Use the ThreadPool if the TaskPool is not available for your platform.