イベントの条件に応じてRx Observableを完了する方法

Question

私には私が制御できないイベントがあり、私にデータを提供しています。 eventArgsは次のようになります。

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

私はこのイベントをIObservableに変換するためにRxを使用します。しかし、私はIsLastItemが真である場合にobservableを完成させたい。

優雅なアイデアはありますか？ 1つの方法は、条件が発生した場合にOnCompleteイベントを設定するために、より多くの制御を持つ対象を介してデータをパイプすることです。

Answer 1

あなたは最後の要素が含まれるようにしたい場合は、一緒にTakeWhileと組み合わせる通常の流れで最後の要素だけでストリームをマージすることができますに。

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

この版画：ここ は、それを証明するために、単純なコンソールアプリケーションです

On Next: test 
On Next: last 
Completed 

UPDATE根本的な観察可能が実際に完了しなかった場合OnCompletedメッセージをsupressedバグがありました 。

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

Answer 2

このようなものをお探しですか？

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