Expert to Expert: Brian Beckman and Erik Meijer - Inside the .NET Reactive Framework (Rx)

Oh my! Oh my! Too cool: Erik, Wes and Brian. More please. I've been wondering what Wes has been up to since his blog went dark.

whoa eric brian and wes. talk about crssing the streams cant wait to watch this one

(also hoping against hope for some natal somewhere in there )

watching it now  i saw erics talk about rx on lang.net (i think it was) and his coin example sounded hilarious to bad they didnt shoot him talking then because it seemd to be alot of visuals besides the slides. did he actually have a bunch of coins and threw them at people?

totally off topic:

if you hold shift while right clicking on a taskbar button in win7, you get the classic context menu with the move/maximize/minimize options. great if your media player is on your other monitor (the TV in my case)

Great video! Really fascinating. I can't wait to see the code - just push it to me

A question: The scope of the  IDisposable object returned by Subscribe() was unclear to me. Erik implies you can put it in a using statement, like a normal IDisposable. But the lexical scope of the IDisposable returned seems much different than the lifetime ("runtime scope"?) of the IObserver you pass in. For example, the using statemetn with IDisposable for files is pretty obvious:

  using(fs = new Stream("somefile.txt"))   
  {     
    // ... do work on fs ...   
  } // fs is closed and disposed. 

But in the IObserver case, I don't think the using statement's scope is appropriate for indicating you don't care about events anymore. Using the autocomplete example above:

using(q.Subscribe(suggestions => ... ))
{
  // ???
} 

What happens in the body of the using statement? All the work seems to occur in the lambda passed to Subscribe. If I understood correctly, as soon as we leave the scope of the using statement, my lambda stops receiving events. Hmm, I bet the object implementing IDisposable has the secret sauce for the reactive framework. Maybe it has a method for testing for conditions and will "yield" until the condition is met? For example if I had another observer which looked for ENTER or TAB on the textbox and indicated that a selection was made, maybe it would let me write this code:

class KeyObserver :  IObserver<...>
 {
    publi c bool SelectionMade()
 }

KeyObserver key = ...;

using(cntrl = q.Subscribe(suggestions => ... ))
{
   if(! key.SelectionMade())
     sleep;
} 
// selection has been made and we stop listening to suggest events

Hopefully that makes some sense. It looks like the beginnings of a way to do reasonable event based programming. Apologies for answering my own question but glad to hear if I'm on track ...

Two other notes:

• Check out the AsyncEnumerator library from Jeffrey Richter. He leveraged enumerators and iterators to do something similar, but I think his solution still has an "inversion of control" problem where you need to write the code to drive the enumerator. A recent article can be found at http://msdn.microsoft.com/en-us/magazine/cc721613.aspx
• Erik mentioned a paper by Phillip Wadler called "The essence of monads" or similar. I think he meant "The essence of functional programming" (http://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.38.9516).

Great stuff, get me this code!

Justin

Be sure to check out Eric's talk at Lang.NET Symposium 2009 (which he alluded to in this video) and Anders Hejlberg's covariance/contravariance in C# 4.0 talk.

Using disposable resources like FileStreams in the deferred execution model of LINQ queries (and closures in general) can be frustrating.  Beware of leaks as well as premature disposals.

Love it guys.

While watching I was thinking about the analogy between IObservable<T> and the CCR's Port<T> and how it sounds like a port abstraction.  Port<T> allows you to monitor a port with various typed-listeners and you can listen for stop, exception, and value results (and also return result back to the port) and also do things like Choice over various ports.  How does IObservable compare/contrast to the port model? 

How does fifoness play into this?  I know with Port, sometimes you have to manually assure fifoness because of the inherent concurrency.

btw - "Please subscribe to this Channel 9 interview..".  You guys jest, but that could be a killer feature.  Being able to post arbitrary Observer code in a c9 reply that could do work for you (i.e. send a tweet, send a mail, etc) based on some code in your observer.  Like a little bot script.  That could be the next big thing in blogs and forums.  High geekness factor.

heavy stuff

im most likely missing something but isnt this very similar to how events work? the event is sort of like the Ioberservable and the delegate you "pass in" with += is like the Iobserver? also, in the code example above you call a method called subscribe, why is that? is that equivalent to the implicit "subscribe" of th e += syntax? in the discussion of event its mentioned that you can sort of leak delegates to the event, but isnt that also true for Iobserveble?

Rx seems very cool but i still feel i havent completly wrapped my brain around it cant wait for more info on this

--edit--

aah, is the "linq" statement above just in essennce just defining whats beeing oberved? so the Iobserver not only knows its observing "something" it also encapsulates what exactly it observes,  in this case the "fragment" in "textbox"

i feel im getting a little closer to understanding this, but i cant wait for some code/examples

--edit2 post rewatch--

hm my original questions seems somewhat trivial, of course it similar to events, but Rx is event on sharksteroids this is what i consider trlu beautiful code, its sooo simple but incredibly powerful.. just like ienumerable

To see if I have grasped the basic concepts, i summarised them. Are they correct?

IEnumeratable & IEnumerater: Over there is some stuff, take something from it and put it over there.

IObservable & IObserver: When you have something don't pass it to me, place over there.

Truly fantastic.. Please make it MS-PL so it's available on Mono..

Thanks..

This is flippin' cool!

My brain just got kicked right in the ambient monads.

Awesome!  Of course no where near the "polishness" or completeness of your work but recentley I implemented a class to follow up the progress of worker threads that fits perfectly pattern of the IObserver interface. 

It's nice to see how my device is validated by your work or that I can now say I've used things with fancy names: "hey! I've used reacting programing"

As usual this is looking terrific (and watching the preview at lang.net this is looking promising... back to watching)
One more thing: using Maybe<T>, you could replace MoveNext: () -> Bool with MoveNext: () -> (); of course the whole thing could be implemented as just Next: () -> Maybe<T> to begin with. That would break command/query separation in a way, so maybe Next: () -> () and Item: () -> Maybe<T> is better.

[Functional] Reactive Programming (FRP). Now Object-Oriented Reactive Programming (OORP)?

In the C# world we have exceptions, so in the context of C# it is appropriate to model errors directly in the IObserver interface right, although one could do away with exceptions as side-effects and model them as types - which would probably be a more disruptive programming model. Hmm...

You can turn an IEumerable into an IObservable by doing something like this:

while(it = MoveNext()) callback.call(it)

Or you can sample every second, or sample when something else happens, etc.

Can you do the reverse with Rx: turn an IObservable into an IEnumerable? If you have coroutines you can do that by setting up a coroutine that subscribes to the IObservable and every time it gets called it yields the element it receives. I've seen this trick used somewhere to turn an collection interface that has only ForEach into a IEnumerable with MoveNext.

Edit: You could also use a thread of course but that would be very expensive.

An intuitive explanation goes like this.

You have a Supply and a Demand. Your job is to get stuff from the Supply to Demand. Now, there are two types of Supplies:

1) PushSupply, they push stuff to you (IObservable)

2) PullSupply, you have to ask them (IEnumerable)

Likewise, there are two types of Demand:

1) PushDemand, you push stuff to the Demand whether they like it or not (user of IObservable)

2) PullDemand, they pull stuff from you (user of IEnumerable, they pull with MoveNext & Current)

If you have a PushSupply and a PushDemand you can just hook them up. They work together naturally. If you have a PullSupply and a PullDemand you can also just hook them up.

A small problem arises if you have a PullSupply and a PushDemand. Now you have to pull stuff from the Supply personally and push it to the Demand. You can do this by pulling and pushing as fast as you can:

while(MoveNext()) callback.Call(Current)

Or you can sample every second, for example.

A bigger problem is bridging the gap between a PushSupply and a PullDemand. The PushSupply pushes stuff in your hands but what can you do if the PullDemand doesn't want it yet? And what do you do if the PullDemand wants new stuff but the PushSupply hash't pushed anything?

One way to do that is this: if the PushSupply wants to give you something but the PullDemand doesn't need it yet, you stop the supply and wait until the demand wants it. If the demand wants something but the supply hasn't pushed yet you stop the demand and wait until the supply gives you an item.

You could also put stuff in a buffer. It depends on the situation: if you have a collection like thing with only a ForEach method you wouldn't want to put the entire collection in the buffer. Especially not with an infinite collection, because you don't have an infinite buffer! You want to execute the producer and consumer in lockstep. Or if they run concurrently with a bounded buffer. Does Rx support these things?

Very interesting video!

right, and they're also composable, unlike events!

chaining/joining events together is kind of painful but its easy with IObservers. and then there is filtering and all that. Also, it seems like IObservers would be great for Ansyc scenarios. i belive someone mentioned that in the comments too  are you doing anything perticular with the TPL team or with the CCR?

you could stretch the sushi analogy even further, events are like the sushi chef making diffrent sushi pieces and handing them to you leaving you to pick out the ones you like. But IObserver is like a waiter into whom you encode what pieces you like and he'll only bring you those. the chef doesnt have to keep track of what you want, he only makes random sushi and you dont have to filter the sushi as its coming in, you tell the waitor once and the stream of sushi comming at you is already filterd. it also works for joining, you tell the waitor that you only want to have your calefornia roll with a salmon piece, and the waitor can bring you those in pairs even if there are two chefs working asynchronosly and you dont have to do any joining in you handler code. (ie, putting the calefornia roll in your mouth and then sitting there, mouth open waiting for the salmon, that'd look silly, and its silly to do it in code too)

is that correct?

--edited post, added sushi--

Very beautiful and simple.

I was so impressed by the simplicity of it all that I tried to do a sample implementation in C#.
Did I get it right?

http://reactivelinq.codeplex.com

This is really great, especially the tie-in with concrete mathematical concepts.  It does, however, make me wonder if this pattern could be expanded upon to handle a "distributed" observer pattern, where remote clients (identified by a URI, not a function pointer) can subscribe to events occurring another machine.  I've worked on projects doing just this in the past (via SOAP/HTTP), but it's a lot of plumbing.  WCF does support the notion of callbacks, but to say their approach is intuitive is a bit of a misnomer (quite ugly if you ask me).

Thoughts?

Pretty cool work. I think your implementation points out a few issues that this approach has to deal with.

1) SpotAgent allows many subscribers; RiskAgent only one -- That says the Subscribe method isn't giving enough information (and the IObservable interface isn't rich enough). I wonder how you could indicate that a given Observable allows many observers vs. one?

2) One problem that Erik points out with IEnumerable is the "subscriber" which calls GetNext() has no control over how long the "publisher" take to return the next value. I think IObservable reverses this problem. That is, the "publisher" (IObservable object) has no control over how long given "subscriber's" (IObserver) OnNext method will take to complete. You can see this issue in the Tick() methods for SpotAgent and RiskAgent.

Does the composobility of IObservable/IObserve help solve this problem?

Looking at your code definitely exposes some issues that I'm sure Erik's team has faced when developing their library.

I didn't see any announcements about this, so forgive me if this is a repost, but has this accidentally (or perhaps intentionally?) shipped with the Silverlight Toolkit?

Just for the fun of it,

I've added an F# implementation of SelectMany and Select for IObservable in F#.

http://reactivelinq.codeplex.com/

holoed

http://fsharpcode.blogspot.com/

rx has apparently made it into the silverlight toolkit

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

Is RxLinq part of the Microsoft Tesla long term initiative ?

it was actually im having som compilation issues when using it with a regular .net project though, most likely because its a silverlight dll. also, i cant wait to hands on some docs, what is the Amb extension method for example? and does Rx use tpl for synchronization? 

questions, questions..

A little brainstorming after having watched the video:

S1 ; S2

In the language of event-driven/reactive paradigms this could be read as: statement 2 is one observer of the state manipulations done by statement 1... now guys hold on to your brains: what happens when there is more than one observer to S1? The whole idea of one statement after another "breaks" as S1 is followed by S2a and at the same time by S2b and S2c etc.etc. S2a, S2b S2c etc all get notfied and start executing at the same time... A program transforms from a list of statements into a tree of statements? Parallel computing? Speculative computing? Quantum computing?

It is deeply (almost disturbing and) fascinating to see how IObserver actually covers the continuation monad, which more or less seams to be the essance of "statement after statement with shared state" programming. One statement tells the next to execute or communicating an error or the end of the program.

Another twist: what happens when one statement S2 can get hold of the handle connecting/making it the succesor of S1 (the IDisposable obj) .... statements reconnecting? Programm-parts actively rewireing based on meta-state? 

How about not using IDisposable but instead an extended "handle" allowing not only to stop the subsription but also to interrupt (for certain time?) or control it otherwise? (have to admit i did not check out the code yet, maybe that is already available)

And the last for today: what other dualities are there waiting to be found?

I would just like to ask, since Erik mentioned that this IObservable captures the essence of a Continuation, does that mean something like call-with-current-continuation could be implemented in C# using the IObservable?

I am not very familliar with C#, and I have never fully grokked call/cc, so bear with me if this question is lacking some basic fundamental understandings of either!

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

Did you forget something?

The Subsrciber to the Observer assumes that the Observer, and the connection to the Observer, are functional during the life of the subscription. If the Observer, or the connection to the Observer, fail, the Subscriber may never know and assume that all is well when it is not. What happens to the Subscriber if the Observer, or the interface to the Observer, fail? How is that handled?

Are other features required so that the Subscriber can periodically verify the health of the Observer and not blindly assume that everything is OK when no events are received ?

Awesome!

Can we please have more of Erik with Brian together.

Let them discuss varinats in type system or other academic stuff. Or mroe real life MSR projets, for example spec#, spec explorer, axum etc.

Maybe someday you can get Yury Gurevich and Tony Hoare together

Thanks Charles, awesome video.

Wow. Were they way off on the covariance and contravariance example earlier in the video. Btw, the array covariance in c# is broken, and it was never legal (or wise) to have a Apple[] have a Fruit[] as the backing storage. Erik was plain wrong on that being safe. Anyhow, a cool vid otherwise.

I think they were implying the work done around co and contravariance in C# 4.0.  Could be wrong though.

Douglas

For what it's worth, a solution to the deferred disposal problem can be found here: Using IDisposables with LINQ. The filestream example would look like this:

IEnumerable<string> lines =
    from fs in new Stream("...").Use()
    from l in fs.ReadLines()
    where l.Length > 256
    select l;

foreach(var l in lines)
{
    // fs will be disposed with the enumerator
}

The same approach can be used with a delegate to defer creation as well as disposal: () => new Stream("..."). A use keyword, a la F#, would be even better.

I was quite happy with the coverage - it was already unusual to mix physics and programming this way - very stimulating although beyond me. One of the best parts was the formulation of co- and contravariance rules via sequent notation, dualization and showing a statement sequence as a series of continuations, if I got that right. Definitely a video that expands horizons. And this stuff about ambient monads. I'm not sure that was really covered anywhere, was it?

I'm wondering if given the excitement about the dual nature, we might be losing part of the intention of IObservable.

First of all, I'm thinking there might be a bit of a mismatch in the duality anyway.  Primarily, IEnumerator and IEnumerable have a many-to-one relationship, whereas IObserver and IObservable havea many-to-many relationship, right?  The deal is that IEnumerator has to have intimate knowledge about the structure of its IEnumerable that the app doesn't necessarily have; they're tightly bound.  Whereas IObserver and IObservable are completely unaware of each others' internals. 

I think the dual of IEnumerable would be something more like ILoadable, where you call ILoadable.GetLoader(), which then allows you to load values into an "array" or whatever, using Loader.MoveNext(bool) and Loader.SetValue(T).  I certainly think the intention of IObservable is more valuable than any ILoadable concept (though ILoadable might be an interesting way to explore inserts and updates), so I wonder if the focus on duality is perhaps the wrong way to go about it.

For example, I'm not sure I get the point of the OnError method.  First of all, isn't there a duality mismatch here anyway—given that IObserver.OnNext(T) could also throw an exception, then to preserve absolute duality going the opposite direction, shouldn't we require that OnError also be a part of IEnumerator?  But beyond that, it just doesn't seem to fit.  Why would the Observable need to tell all Observers that there was some exception downstream?  Just like an iterator doesn't need to know that there's an exception occurring in the app that uses it.  So I think the focus on preserving duality really mucks up the intention of IObservable in this case.

Also, I think if duality was that important to the concept of IObservable and IEnumerable, then the dual of (for example) IList would be even more powerful.  But wouldn't the dual of (for example) IList.Set(int, T) be IListDual.Get(out int, out T)?  Is that useful in some fashion?  I don't see anything, though I'd love to hear from anyone with an idea how it could be powerful.

Covariant/contravariant! 

Help! Someone call Eric Lippert

Apple[] as
Fruit[] fs = as
Fruit f = fs[0]
fs[1] = new Banana

In C#, the last line will throw ArrayTypeMismatchException. :

Hi Guys,

Can someone explain this:

var values = (new[] { "test 1", "test 2", "test 3" }).ToObservable();

values.ToEnumerable().AsParallel().ForAll(Console.WriteLine);

ForAll method will never return. Why is this happening (I have November rx release for .net 3.5)?

Thank you.

I'd be curious to know any of your thoughts on my own implementation of IObservable, which I had to rename to INotify since Microsoft is rolling the Reactive Framework out and including it in .NET version 4. I too 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 if I have a screw loose or if I'm on the right path...

What is the difference between a publish/subscribe and observer/subject architectures?

Just started watching this. About co/contravariance w.r.t. subtyping: producer types are covariant, and consumer types are contravariant to subtyping. Apple is a subtype of Fruit: wherever Fruit is used, Apple can be used too: A <: F. That is because any quality Fruit has, Apple has too.

Array read is a producer: give it an Int and it gives you an Apple. Now, A <: F => (Int->A) <: (Int->F). Wherever we use (Int->F), it gives us Fruits, so we must be prepared to handle Fruits there. But it's OK to use (Int->A) instead 'coz it'll give us Apples, which can go instead of Fruits always.

With consumers it's the other way around. A<:F => (F->Int) <: (A->Int). Wherever we use Apples consumer, (A->Int), it is prepared to use Apples. So it means we must be supplying it with Apples. So we can use fruits consumer (F->Int) in its place, because this fruits consumer can always use Apples instead of Fruits.

Graphically, we can envision the producers/consumers as pipes with a certain diameter - small for apples, bigger for fruits. There are many more fruits than there are apples - bananas too etc. Now, for a wide inlet of fruits consumer (us, in the 1st case), it can just as well handle all input from a narrower outlet of an apples producer. In the 2nd case, the narrow outlet of apples producer (us) can just as well go into a wider inlet of fruits consumer in general.

Array read is a producer; array write is a consumer. Does this make sense?

EDIT: and in general, covariant, when translated from Latin I gess, just means "changing with the direction of change", and contravariant means "changing in the opposite direction of change". Say we do something that enlarges one thing, and another one grows too, than it is covariant to the first w.r.t. our action. If it consistently shrinks, we say it's contravariant to the first w.r.t. our action.

For example, imagine you travel from city A to city B in one straight line. The further you go along the route, the further you are from the city A, so your distance from city A is covariant with the distance you travel in the car. But the distance to city B is contravariant to it - it grows smaller as you travel. I guess a "travel operator" in its covariant form will reflect the enlargement of the distance as measured from the origin point. But in its contravariant form it will show the shrinkage of distance as measured to the destination.

Or take rotation: say we have a point on a plane, in some coordinate system. Now let's apply rotation to that plane. Point's coordinates will change WITH the plane's rotation, or covariantly. But what if we apply rotation to the coordinate system instead? Now the point's coordinates will change _in the opposite direction_ to the rotation, or contravariantly. That's where these terms come into physics from.