Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How to complete a Rx Observable depending on a condition in a event

I have a event that I'm not in control of which provides me with data. The eventArgs looks something like this:

class MyEventArg {
  bool IsLastItem {get;}
  Data DataItem {get;}
}

I use Rx to convert this event to an IObservable. But I want to complete the observable if IsLastItem is true.

Any elegant ideas? One way would be to pipe the data through a subject that I have more control over to set the OnComplete event if the condition occurs...

like image 387
lukebuehler Avatar asked Nov 14 '11 18:11

lukebuehler


3 Answers

If you want the last element to be included you can merge a stream with only the last element together with the regular stream combined with TakeWhile. Here is a simple console app to prove it:

var subject = new List<string>
{                            
"test",
"last"
}.ToObservable();

var my = subject
            .Where(x => x == "last").Take(1)
            .Merge(subject.TakeWhile(x => x != "last"));

my.Subscribe(
    o => Console.WriteLine("On Next: " + o), 
    () => Console.WriteLine("Completed"));

Console.ReadLine();

This prints:

On Next: test
On Next: last
Completed

UPDATE There was a bug that supressed the OnCompleted message if the underlying Observable didn't actually complete. I corrected the code to ensure OnCompleted gets called

And if you want to avoid subscribing to the underlying sequence multiple times for cold observables you can refactor the code like this:

var my = subject.Publish(p => p
            .Where(x => x == "last").Take(1)
            .Merge(p.TakeWhile(x => x != "last")));
like image 187
Christoph Avatar answered Oct 24 '22 13:10

Christoph


public static IObservable<TSource> TakeWhileInclusive<TSource>(
        this IObservable<TSource> source, Func<TSource, bool> predicate)
{
    return Observable
        .Create<TSource>(o => source.Subscribe(x =>
                                                   {
                                                       o.OnNext(x);
                                                       if (!predicate(x))
                                                           o.OnCompleted();
                                                   },
                                               o.OnError,
                                               o.OnCompleted
                                  ));
}
like image 25
Sergey Aldoukhov Avatar answered Oct 24 '22 15:10

Sergey Aldoukhov


Are you looking for something like this?

IObservable<MyEventArg> result =
    myEventArgObservable.TakeWhile(arg => !arg.IsLastItem);
like image 41
dtb Avatar answered Oct 24 '22 15:10

dtb