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

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

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

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

任何优雅的想法？一种方法是通过我可以控制的主题来管道数据,以便在条件发生时设置OnComplete事件…

解决方法

如果要包含最后一个元素,则可以将流与仅包含最后一个元素以及与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
如果底层的Observable实际上没有完成,则有一个错误压制了OnCompleted消息.我更正了代码以确保调用OnCompleted

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

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