Perform action after all Observers complete on Observable sequence

0

votes

I have some observable sequence, for example:

var period = TimeSpan.FromSeconds(0.5);
var observable = Observable
    .Interval(period)
    .Publish()
    .RefCount();

I want to perform some hard computations for elements of this sequence on a background thread and also perform some final action when all computations are done. So I want something like this:

observable.ObserveOn(Scheduler.Default).Subscribe(i => ComplexComputation1(i));
observable.ObserveOn(Scheduler.Default).Subscribe(i => ComplexComputation2(i));
// next observer must be called only after ComplexComputation1/2 complete on input i
observable.Subscribe(i => FinalAction(i));

Can I do this in Rx? Or maybe this violates some principles of reactive programming and I should use another approach in such situation?

c#system.reactiveobservable

2

votes

It is very dangerous to have computationally ordered sequences in Reactive patterns.

One thing you can do is have each complex calculation emit an event after it is finished. Then you can have a consuming observer who will perform his calculation once he receives messages that the previous steps are finished.

Another possible solution is to create a concrete sequence block that gets fired regularly. This reduces the parallelizability of the solution.

observable.ObserveOn(Scheduler.Default).Subscribe(i => 
{     
    ComplexComputation1(i));
    ComplexComputation2(i));
    FinalAction(i);
}

1

votes

To test this out I created the following methods to help illustrate the sequence of events:

public void ComplexComputation1(long i)
{
    Console.WriteLine("Begin ComplexComputation1");
    Thread.Sleep(100);
    Console.WriteLine("End ComplexComputation1");
}

public void ComplexComputation2(long i)
{
    Console.WriteLine("Begin ComplexComputation2");
    Thread.Sleep(100);
    Console.WriteLine("End ComplexComputation2");
}

public void FinalAction(long i)
{
    Console.WriteLine("Begin FinalAction");
    Thread.Sleep(100);
    Console.WriteLine("End FinalAction");
}

Your original code ran like this:

Begin FinalAction
Begin ComplexComputation1
Begin ComplexComputation2
End ComplexComputation2
End FinalAction
End ComplexComputation1
Begin FinalAction
Begin ComplexComputation1
Begin ComplexComputation2
End FinalAction
End ComplexComputation2
End ComplexComputation1
Begin FinalAction
Begin ComplexComputation1
Begin ComplexComputation2
End ComplexComputation2
End ComplexComputation1
End FinalAction
...

It's easy to enforce the code to run in sequence on a single background thread. Just use an EventLoopScheduler.

var els = new EventLoopScheduler();

observable.ObserveOn(els).Subscribe(i => ComplexComputation1(i));
observable.ObserveOn(els).Subscribe(i => ComplexComputation2(i));
// next observer must be called only after ComplexComputation1/2 complete on input i
observable.ObserveOn(els).Subscribe(i => FinalAction(i));

That gives:

Begin ComplexComputation1
End ComplexComputation1
Begin ComplexComputation2
End ComplexComputation2
Begin FinalAction
End FinalAction
Begin ComplexComputation1
End ComplexComputation1
Begin ComplexComputation2
End ComplexComputation2
Begin FinalAction
End FinalAction
Begin ComplexComputation1
End ComplexComputation1
Begin ComplexComputation2
End ComplexComputation2
Begin FinalAction
End FinalAction

But as soon as you introduce Scheduler.Default this doesn't work.

The more-or-less simple option is to do this:

var cc1s = observable.ObserveOn(Scheduler.Default).Select(i => { ComplexComputation1(i); return Unit.Default; });
var cc2s = observable.ObserveOn(Scheduler.Default).Select(i => { ComplexComputation2(i); return Unit.Default; });

observable.Zip(cc1s.Zip(cc2s, (cc1, cc2) => Unit.Default), (i, cc) => i).Subscribe(i => FinalAction(i));

That works as expected.

You get a nice sequence like this:

Begin ComplexComputation1
Begin ComplexComputation2
End ComplexComputation1
End ComplexComputation2
Begin FinalAction
End FinalAction
Begin ComplexComputation2
Begin ComplexComputation1
End ComplexComputation2
End ComplexComputation1
Begin FinalAction
End FinalAction
Begin ComplexComputation1
Begin ComplexComputation2
End ComplexComputation2
End ComplexComputation1
Begin FinalAction
End FinalAction

0

votes

This seems like a simple case of composition of a nested observable being flattened (SelectMany/Merge/Concat) and Zip

Here I have taken the liberty to assume the Long Running methods return a Task. However if they dont, then the slow blocking synchronous method could instead be wrapped with Observable.Start(()=>ComplexComputation1(x)).

void Main()
{
    var period = TimeSpan.FromSeconds(0.5);
    var observable = Observable
        .Interval(period)
        .Publish()
        .RefCount();

    var a = observable.Select(i => ComplexComputation1(i).ToObservable())
                .Concat();
    var b = observable.Select(i => ComplexComputation2(i).ToObservable())
                .Concat();

    a.Zip(b, Tuple.Create)
        .Subscribe(pair => FinalAction(pair.Item1, pair.Item2));
}

// Define other methods and classes here
Random rnd = new Random();
private async Task<long> ComplexComputation1(long i)
{
    await Task.Delay(rnd.Next(50, 1000));
    return i;
}
private async Task<long> ComplexComputation2(long i)
{
    await Task.Delay(rnd.Next(50, 1000));
    return i;
}

private void FinalAction(long a, long b)
{

}

Perform action after all Observers complete on Observable sequence

3 Answers