我有一个事件,我无法控制它,但它会提供给我一些数据。事件参数看起来像这样:
class MyEventArg {
bool IsLastItem {get;}
Data DataItem {get;}
}
我使用 Rx 来将此事件转换为 IObservable。但是如果 IsLastItem 为 true,我希望完成这个可观察对象。
有没有更优雅的想法?一种方法是通过一个 subject 来传递数据,以便更好地控制它,从而在条件出现时设置 OnComplete 事件...
我有一个事件,我无法控制它,但它会提供给我一些数据。事件参数看起来像这样:
class MyEventArg {
bool IsLastItem {get;}
Data DataItem {get;}
}
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")));
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
));
}
IObservable<MyEventArg> result =
myEventArgObservable.TakeWhile(arg => !arg.IsLastItem);