如何有条件地停止 observable 的定时间隔序列？

【问题标题】：how to conditionally stop an observable's timed interval sequence?如何有条件地停止 observable 的定时间隔序列？
【发布时间】：2020-03-12 16:51:36
【问题描述】：

这是每秒运行的代码，用于按顺序输出一个数字：

IObservable<DateTimeOffset> timestamps =
    Observable.Interval(TimeSpan.FromSeconds(1))
    .Timestamp()
    .Where(x => x.Value % 2 == 0)
    .Select(x => x.Timestamp);
timestamps.Subscribe(x => Console.WriteLine(x));

它有效。我的问题是，我希望 Observable 会生成这个时间戳并在满足某些条件时停止，例如

if (x==100) then stop

有没有办法通过 Rx/Linq 实现这一点？

非常感谢。

【问题讨论】：

标签： c# conditional-statements observable system.reactive intervals

【解决方案1】：

“停止”是通知OnComplete。 “投掷”是通知OnError。有操作员可以推送这两种类型的通知。

您还可以使用TakeWhile 运算符，IEnumerable 中也提供该运算符，它们的行为非常相似。

Observable.Interval(TimeSpan.FromSeconds(0.01))
.Where(x => x % 2 == 0)
.TakeWhile(x => x != 100)
.Timestamp()
.Select(x => x.Timestamp);

对于与您的流异步的通知，有...Until 运算符，例如TakeUntil、SkipUntil。例如，在 100 秒后停止 observable。

Observable.Interval(TimeSpan.FromSeconds(0.01))
.Where(x => x % 2 == 0)
.TakeUntil(Observable.Timer(TimeSpan.FromSeconds(100)))
.Timestamp()
.Select(x => x.Timestamp);

【讨论】：

绝妙的答案！