Kim Hamilton and Wes Dyer: Inside .NET Rx and IObservable/IObserver in the BCL (VS 2010)

So I take it this is not in VS 2010 beta 1, but may be in beta 2? Downloading...

A.W.E.S.O.M.E. I don't suppose there's any rough estimate of Beta 2's arrival Charles? Is breath holding advised?

http://www.paulbatum.com/2009/07/reacting-to-reactive-framework-part-5.html

http://evain.net/blog/articles/2009/07/30/rebasing-system-reactive-to-the-net-clr

In your first C# sample...

var o = Observable.Return(1); 
o.Subscribe( x => Console.WriteLine("OnNext({0})", x), ex => Console.WriteLine("Error()"), () => Console.WriteLine("Completed()") );

When you say 2010 BCL, you mean .NET 4 right?

I guess the implementation of the Observable.Subscribe method that is coming out of Observable.Return just calls OnNext.

So in fact by subscribing you're calling yourself back with the value 1. 

Observable.Return<T>(T value) = new IObservable { Subscribe(observer) = observer.OnNext(value) }

Very cool.

Wes really is a rockstar developer.

Wes, please resume blogging, we miss you.

Nice. And what a funny surprise with Beckman, trained in the art {M[onad],F[unction],C[omposition]}-fu to join in  Now to wait for the arrival of the next beta.

I'm curious how GUI programming will evolve with the arrival of Rx.

Well the Obserable.Return function doesnt just return any old IObserable.  The one it returns calls OnNext the moment after someone subscribes (as it doesnt need to wait, it already knows what it should return).

Interesting stuff, yet another reason to look forward to the RTM release!

Out of curiosity, like you mentioned developers tend to use IEnumerable over arrays, do you think developers will have the tendancy to use this over events in C# .NET? Will any of the controls we use today be extended with IObservable implementations?

Interesting talk! But why you guys call it OnError() ? I think OnException() would be more appropriate as there's quite a different meaning to both words.

I was thinking about the new IObservable interface.  At first I thought great, brilliant, amazing, revolutionary.  But then I thought about how you could take an IEnumerable and make it an IObservable, and the reverse.  For instance you can just block the MoveNext until the next event comes in, or you can just call getnext and then IObservable.OnNext(Current) to switch it the other way.  So in the end what makes IObservable any better or different?  Well the IEnumerable has a slight assumption that when you do a GetNext/Current it will be fairly fast(as it is synchronous), the IObservable makes no such assumption.  If you want to continue computation while you wait for the GetNext to finish you better do that on a different thread, so there is thread overhead.  In the end I would say that IObservable is STRICTLY better than IEnumerable.  Anything IEnumerable can do IObservable can do.  For instance: foreach(type a in IEnumerable){ write(a)} is the same as bool end=true;IObservable.Subscribe(a=>write(a),a=>throw a,()=>end=false;);while(end){};  I bet you could even have the same syntactic suger of the foreach work for IObservable.  The only difference is going the other way where the MoveNext blocks would need to be done on a separate thread so IEnumerable is never better (and sometimes worse) then IObserable.

In the end why don’t you just convert ALL IEnumerable’s over to the IObservable pattern (as well as all events)? (I know that you have backwords compatibility issues with this, but other then that)

cool stuff (shorter than usual though) The brian cameo in the end was epic

events since .net2 has been reffered to in the docs as "first class citizens" but they cant be used like the  IEvent<T> mentioned in the video. do you consider the events that exsist in .net today first class still? using Rx with exsisting event is a bit of a hassle precisly because you cant pass them around, is that something you might change in .net4?

the relation between tasks and iobservable is kinda interseting because they sort of overlap.. to bad will tasks wont implement IObservable  seems like a good fit imo.. there are some problems i guess, does it call OnNext when the progress changed or when its done for example. extension methods might be a good way to go but they relly beg to be integrated well an interview with wes and joe duffy would be really interesting waiting for dev11 seems a bit harsh though :/

also, why is AnonomusObserver internal? are the Subscribe overloads/Select the replacement for having it public? it seems like it would be a very useful thing to have when writing integration with Task or my own stuff...

and why is Combine internal? i think there is a BinaryObservable in there somewhere thats also internal.. please dont repeat what you did with TreeVisitorBase (i think its called), the base class for tree visitors used in linq (that is finaly public in .net4, yay ). Having to rewrite that stuff in every project is such a pain.. im not saying thats what you've done but please, please dont make stuff internal lightly

Kim mentioned that only the interfaces are in the bcl, not the extension methods? is this really correct or did i misunderstand? Rx is great but what makes is super powerful are those extension methods :O surely the entire System.Reactive.dll will be in .net4?

well stricktly speaking you are right all IEnumerables can be expressed as IObservables and vice versa. i think thats part of the mathematical dual thing. (atleast i think so, i dont have any formal proof on hand )

IObservable really shines when you have a bunch of things you want to coordinate. consider a keyboard shortcut. to implement that you'd typically have a keyDown event handler and then store the keys that are currently down and look into a list of your keyboard shortcuts. this however means shared state (= generally speaking, bad ) with Rx however you can write a linq statement that manages this.

also, with Rx you get a new "listener" for your keyboard shortcut, something you dont get with just events. with events you have to manage the dispatch of when your shortcut is pressed, with Rx you get an object that you can subscribe to

check out these blog posts for more info (also where i snagged this example)

http://themechanicalbride.blogspot.com/2009/07/developing-with-rx-part-1-extension.html

http://themechanicalbride.blogspot.com/2009/07/developing-with-rx-part-2-converting.html

i dont think Ienumerable is dead though, its more a question of what makes most sense in each case, sometimes we're pulling data and sometimes we're getting it thrown at us, sometimes we want to block while waiting on data/computations and sometimes we dont

you can actually start playing with Rx right now check out this blog for details:

http://themechanicalbride.blogspot.com/2009/07/introducing-rx-linq-to-events.html

also contains lots of good info on why Rx is useful in general

"sometimes we want to block while waiting on data/computations and sometimes we dont"

well IEnumerable is basicly the creator of the object expressing you should (or must if you dont split off a diffrent thread), block while you wait for the result.  IObservable says you "may" block or not block as you wish.  The nice thing about IEnumerable is it expresses that the author doesnt expect it to take too long to get the next value, although even this kind of assumption can be broken slightly.  Look at Linq to Entities or Linq to SQL, both of these take a long time to get all thier values the first time.  Why should the program have to block while it waits for these to finish?  I think IEnumerable will still probably exist for some "in-memory" very quick things, that blocking doesnt really matter on, but I wish there was a better way for authors to express that they dont expect it to take long to get the values then use a totaly diffrent type.

Basicly IEnumerable is sync with a method call to get the result, and IObserable is async with a callback to get the result.

"IObservable says you "may" block or not block as you wish"

well yeah but if you want to block waiting for an observable, thats something you have to sort out your self. there isnt really anything to block on since you dont call the IObserver to get the values, it calls you.

what you say about IEnumberable taking a long time to get the values is true in some cases but in some sort of general way its not because IEnumerable is lazy. IObservable is sort of the opposite, its hyperactive it keeps throwing values at you at its own pace. this could potentially mean that some work is done in vain, your program might throw away the data because its not ready to handle it or the data have become outdated. the lazyness of IEnumerable limits that behavior

"why should the program wait"

i dont think its all that diffrent comparing IEnumerables and IObserables that way though.. neither are have any threading baked in. Even if IObservables are used the program have to wait for the database at some point  

i still think it comes down to preference and convenience in the end you can really use either one, in some cases is really hard to use one or the other and in most cases the devide isnt as large.

IEnumerables long lost twin has finally made it into .net, making it twice as powerful   (sans the extension methods it seems :/)

If you're interested in a nice challenge with Rx, go over to my blog and try solving the tripple-click puzzle: 

This reply is a bit late perhaps, but I don't quite understand this.

Does Return really call OnNext after someone subscribes, or calls it OnNext during the call to Subscribe? If I call Subscribe, I get back an IDisposable, but do I get the change to call Dispose before the Observable starts pushing values?

In other words, what if I want this method:

IObservable<T> ToObservable<T>(this IEnumerable<T> xs) { return new IObservable<T>() { IDisposable Subscribe(IObserver<T> obs) { foreach (var x in xs) obs.OnNext(x); obs.OnDone(); return someDisposable(); // Don't know how to implement
 } } }

Can I unsubscribe from the IObservable before or during the series of OnNext calls?

If not, how would this be implemented in a way that does allow for me to unsubscribe, should I explicitly spawn a new thread?

Yes, you can unsubscribe. You don't need to use another thread, but given the ToObservable() method you do.

public static IObservable<T> ToObservable<T>(this IEnumerable<T> xs) {
 return new Observable<T> () { Enumerable = xs };
}

Here's the internal class needed to handle subscription:

internal sealed class Observable<T> : IObservable<T> {

private readonly UnSubscribe _unsubscribe = new UnSubscribe (); 
public IEnumerable<T> Enumerable { get; set; } 
public IObserver<T> Observer { get; set; } 
public IDisposable Subscribe(IObserver<T> o) { 
Observer = o; 
ThreadPool.QueueUserWorkItem(Worker); 
return _unsubscribe; 
} 
private void Worker (object state) { 
foreach (T item in Enumerable) {
 if (_unsubscribe.Cancel)
 return; 
Observer.OnNext (item); 
}
 Observer.OnCompleted ();
 }
}

And here's the IDisposable internal class that flags when the subscriber is no longer interested:

internal sealed class UnSubscribe : IDisposable {
 public bool Cancel { get; set; }
 public void Dispose() {
 Cancel = true;
 } 
}

I guess I'm not as familiar with the .NET BCL as I thought, I didn't know about the ThreadPool class.

But from what I've read, it might still schedule the work on another thread (or the current thread if it becomes idle first). So I guess I'm implicitly using another thread instead of explicitly creating my own.

Since the Observer.Return method doesn't use the ThreadPool class (at least not in the Silverlight binary), I guess that means that, that one isn't "unsubscripable".

Because I already wrote my own implementation of IObservable with what appears to be a very different solution, I'd be curious to know what I can do to merge it with the Reactive Framework. My solution was developed because I was frustrated with managing weak event references as well as the implementation of events in WPF and now Silverlight using INotifyPropertyChanged versus DependencyProperty. Let me know...

I like Rx very much but before I invest 41 minutes of my time let me please ask this:
Is this video still up to date? Rx has undergone some changes since the production.