Synchronize the order of multiple observables

.net, c#, system.reactive

Solution

This does what you want using observable joins:

    using System.Reactive;
    using System.Reactive.Subjects;
    using System.Reactive.Linq;

    var a = new Subject<int>();
    var b = new Subject<int>();
    var c = new Subject<int>();

    var test = 
        Observable.When(
            a.And(b).And(c).Then((a1, b1, c1) => new int[] { a1, b1, c1 })
        ).SelectMany(arr => arr.ToObservable());

    var sub = test.Subscribe(val => Console.Write("{0} ", val));

    a.OnNext(1);
    b.OnNext(2);
    c.OnNext(3);

    a.OnNext(1);
    c.OnNext(3);
    b.OnNext(2);

    c.OnNext(3);
    a.OnNext(1);
    b.OnNext(2);

    Console.ReadKey();
    sub.Dispose();

The output is:

1 2 3 1 2 3 1 2 3

Notice however that:

This doesn't provide an extension method that allows you to inject a particular observable in some order. You would need to build the plan up using `And` (which can be done piecemeal), and then eventually execute it. The 'build up' order will be the output order.

This doesn't look at the actual window creation. However, there's no reason that couldn't be injected into the observable chain somewhere.

This solution will fire once every time all three observables have a value. You may need to set it up and tear it down at the right points to get the desired behaviour.

Problem

I have multiple (2-10) hot observables, each fires one event in a given time window but not in order. Lets say I have 3 subjects, and to illustrate it they fire in following arbitrary orders: ``` Subjects: sA, sB, sC Order of time window 1: sB, then sA, then sC Order of time window 2: sA, then sC, then sB Order of time window 3: sA, then sB, then sC etc. ``` So if I would do a merge I would get following stream: sB, sA, sC, sA, sC, sB, sA, sB, sC, etc. Now, here's my question: I'd like to force the order of events within each time window. ``` 1. sA 2. sB 3. sC ``` Which would result in following merged stream: sA, sB, sC, sA, sB, sC, sA, sB, sC, etc. The window time itself is not known, but since each event fires exactly once in each window we can assume that once all subjects have fired the windows is closed and a new one is opened. Any Ideas how to do this elegantly? Extended question (not as important): the observables are independent of each other so i need to introduce a common static method where I can synchronize the order of each observable. ``` public static IObservable<T> SynchronizeOrder<T>(IObservable<T> source, int order) { //return synchronized source } ``` Update: Something that wasn't clear in my original question was that the observables can be of different event types (that's why I didn't use a specific event data type in my example) and consequently I do not want to merge the ordered observables. I want to have a mechanism where I can ensure that independent observables are being synchronized in their order. ``` myEventStream .ObserveOn(TaskFactory) .DoSomeExpensiveComputation() .Order(2) //doesn't have to be an extension method .Subscibe(computationResult=> sharedRessourceWhereOrderMatters.Update(computationResult)) ```

Original source