如何根据事件中的条件完成Rx Observable

Question

我有一个我无法控制的事件为我提供了数据.eventArgs看起来像这样:

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

我使用Rx将此事件转换为IObservable.但是如果IsLastItem为真,我想完成observable.

任何优雅的想法？一种方法是通过我可以控制的主题来管道数据,以便在条件发生时设置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

更新 如果基础Observable实际上没有完成,则有一个错误压制了OnCompleted消息.我更正了代码以确保OnCompleted被调用

如果你想避免为冷可观察量多次订阅底层序列,你可以像这样重构代码:

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