Guide to System.Reactive.Joins

Question

I'm looking for an introduction/ some documentation of System.Reactive.Joins, which includes the Pattern, Plan, QueryablePattern and QueryablePlan classes. Google doesn't turn up anything ("System.Reactive.Joins"), MSDN has nothing, there are no samples here, and the excellent resources from this question do not cover this namespace.

Does anyone have some pointers?

Answer 1

Found an excellent SO question that shows the usage, but to me the overall purpose of Plan and Pattern is to create a compositional unit of observable as opposed to a composed observable. Semantics, I know, but to me it seems a little easier to use this syntax then the various other "Join" methods. It allows you to separate the join from the projection altogether, so you can store intermediate plans and compose them with other observables whenever you want.

For example:

// Suppose we have observables o1, o2, ..., o9. 
// All IObservable<int>.

var o1and2 = o1.And(o2);  // Store this bad boy for later use. Pattern<int, int>

var o5and6and9 = o5
                .And(o6)
                .And(o9)
                .Then((t1, t2, t3) => t1 + t2 + t3);  // Plan<int>

var o3and7 = o3
            .And(o7)
            .Then((t1, t2) => string.Format("Result: {0}", t1 + t2)); // Plan<string>

var o12ando8and6 = o1and2
                  .And(o8)
                  .And(o6)
                  .Then((t1, t2, t3, t4) => ((decimal) t1, t2, t3.ToString(), t4));   
                  // Plan<(decimal, int, string, int)>

// "When" groups similar results together.
// It will fire when any of the Patterns give a result.

var obs1 = Observable
          .When(o1and2.Then((t1,t2) => t1+t2), o5and6and9); // IObservable<int>

var obs2 = Observable.When(o3and7);       // IObservable<string>
var obs3 = Observable.When(o12ando8and6); // IObservable<(decimal, int, string,int)>

SO Article: Reactive Extensions for .NET (Rx): Take action once all events are completed

Also, found an RX document that actually helped in understanding HOW to use this: http://www.clipcode.net/mentoring/RxReferenceLibrary.pdf [dead]

Answer 2

This is the only thing I found: Join Patterns in Rx. I would also like to see other resources about these topics.

Answer 3

Reactive Extensions for .NET (Rx) ^{blogs.msdn.com}

... then waits for the first two of three results to return using a join pattern.

How to join multiple IObservable sequences? ^{stackoverflow.com}

How about using the new Join operator in v.2838 ...

Introduction to the Reactive Extensions for JavaScript – New Release and Joins ^{weblogs.asp.net}

... Reactive Extensions for JavaScript which includes many of the changes I’ve been talking about lately including the third-party library integration, aggregates which I covered in the previous posts, and joins which is the subject of today’s post.

System.Reactive.Joins Namespace ^{msdn.microsoft.com}

Microsoft's definition of the System.Reactive.Joins Namespace.

Answer 4

Other than the example I mentioned here (now updated to use Observable.When instead of Observable.Join), the general notion to me seems to be that you'd use When and join patterns when you want to compose the events themselves, rather than the contents of the events. So, I'd use the other query operators, like SelectMany, when I want to do something connected with the data being observed. When I want to do something purely in response to which events, and which patterns of events are firing in a particular order, then I would use join patterns. Now, that's not to say you can't combine the other query operators, say, Where and Select to filter and project out some values from an IObservable, which will later be used in a join pattern, but I think that's a good general way to look at it.

I did try to make some examples but I haven't really come up with anything better ... just more complicated cases of the same notion. You want to group related event patterns which may have complex relationships that make sense at the event level and not at the content of the IObservable.

Answer 5

I cannot find much either. But I had some success with my own trying.

The cotacha is: The type parameters of the delayObservable are not of any meaning. They are just declared to have a don't care return value. Seems (*I dont know) like the purpose of that IObservable<> only is to listen for the OnCompleted event.

Therefore I used in my experiment this Observable:

IObservable<long> timingobs = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(1);

If you find a simpler solution for a quickly ending Observable, let me know.

The Result differs, if you use different amounts (like 600) of milli seconds.

Code:

            var apples = Observable.Create<Apple>(o => Task.Run(() =>
            {
                o.OnNext(Apple.boskop);
                Thread.Sleep(500);
                o.OnNext(Apple.boskop);
                o.OnNext(Apple.granny);
                Thread.Sleep(500);
                o.OnNext(Apple.pink);
                o.OnNext(Apple.pink);
                Thread.Sleep(500);
                o.OnNext(Apple.granny);
                o.OnCompleted();
                return Disposable.Empty;
            }));

            var pears = Observable.Create<Pear>(o => Task.Run(()=>
            {
                o.OnNext(Pear.AA);
                Thread.Sleep(300);
                o.OnNext(Pear.BB);
                o.OnNext(Pear.CC);
                Thread.Sleep(300);
                o.OnNext(Pear.BB);
                Thread.Sleep(300);
                o.OnNext(Pear.DD);
                o.OnCompleted();
                return Disposable.Empty;
            }));



            IObservable<long> timingobs = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(1);
            pears.Join(apples, a => timingobs, p => timingobs, (pear, apple) => $"{apple} + {pear}")
                .Subscribe(Console.WriteLine);

            Console.ReadKey();

Result (100ms):

boskop + AA
pink + DD
pink + DD

Result (600ms):

boskop + AA
boskop + BB
boskop + CC
boskop + AA
boskop + BB
boskop + CC
granny + AA
granny + BB
granny + CC
boskop + BB
boskop + BB
granny + BB
boskop + DD
granny + DD
pink + BB
pink + DD
pink + BB
pink + DD
granny + DD

Answer 6

It's not a bad ide to use Reflector (or similar tool),
for now, going through the source code is invaluable tool for Rx. You can see what it implements, how - and compare to other standard Rx patterns.
Also, maybe you could post a problem, what you're trying to solve - and then go from there (and others might have more ideas).
EDIT: There is no source code for Rx as such. However even the Rx people recommend analyzing the library by disassembling it using Reflector or similar tools. E.g. you can use ILSpy, free, download Rx-Providers Nuget or you have that I guess, load the System.Reactive.Providers.dll and System.Reactive.dll into it and enjoy:). It's not perfect nor official nor it compiles, easily at least - but you can get a good sense of how things work, and it helped me a great deal, invaluable!