Come completare un Rx Osservabile a seconda di una condizione in un evento

Question

Ho un evento che non ho il controllo di cui mi fornisce i dati. I eventArgs simile a questa:

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

Io uso Rx per convertire questo evento per un IObservable. Ma voglio completare l'osservabile se IsLastItem è vero.

Qualche idea elegante? Un modo potrebbe essere quello di pipa i dati attraverso un argomento che ho più controllo sulle impostare l'evento OnComplete se si verifica la condizione ...

Answer 1

Se si desidera che l'ultimo elemento da inserire è possibile unire un ruscello con solo l'ultimo elemento insieme al regolare flusso combinato con TakeWhile. Ecco una semplice console app per dimostrarlo:

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(); 

Questo stampa:

On Next: test 
On Next: last 
Completed 

UPDATE C'era un bug che soppressa il messaggio OnCompleted se osservabile sottostante non ha effettivamente completa . Ho corretto il codice per garantire OnCompleted viene chiamato

E se si vuole evitare la sottoscrizione alla sequenza sottostante più volte per osservabili a freddo si può refactoring del codice come questo:

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

Answer 2

Siete alla ricerca di qualcosa di simile?

IObservable<MyEventArg> result = 
    myEventArgObservable.TakeWhile(arg => !arg.IsLastItem); 

Answer 3

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 
           )); 
}