Cómo completar un Rx Observable dependiendo de una condición en un evento

Question

Tengo un evento del que no tengo control y que me proporciona datos. El eventArgs se parece a esto:

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

Utilizo Rx para convertir este evento en un IObservable. Pero quiero completar el observable si IsLastItem es verdadero.

¿Alguna idea elegante? Una forma sería canalizar los datos a través de un tema sobre el que tengo más control para establecer el evento OnComplete si se produce la condición ...

Answer 1

Si desea que el último elemento que se incluirán puede combinar una corriente con sólo el último elemento junto con el flujo regular combinado con TakeWhile. Aquí es una aplicación de consola sencilla para probarlo:

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

Esta impresora:

On Next: test 
On Next: last 
Completed 

ACTUALIZACIÓN Hubo un error que el mensaje suprimida OnCompleted si el subyacente observable en realidad no completa . Me corregido el código para garantizar OnCompleted obtiene la llamada

Y si se quiere evitar la suscripción de la secuencia subyacente varias veces para observaciones en frío se puede refactorizar el código como el siguiente:

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

Answer 2

¿Estás buscando algo como esto?

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