在Rx中以重复序列跟踪完成情况

Question

你能告诉我是否有办法跟踪重复序列中当前重复的完成情况吗？我的最终目标是跟踪当前的计数和完成重复的速率，一旦我有了基础，我就可以这样做。谢谢你的帮助。

sequence.Repeat().Publish().RefCount()

我想在上面的结构中捕捉"序列"的完成。

Answer 1

使用Materialize将T的可观测值转换为Notification<T>的可观测结果。这允许您像处理普通序列值一样处理值、完成和错误。然后你可以过滤掉完成的内容。然而，我不太明白你打算对他们做什么。。。

var completions = sequence
    .Materialize()
    .Where(notification => notification.Kind == NotificationKind.OnCompleted)
    .Repeat();

Answer 2

Daniel的答案是有效的，尽管它会隐藏流错误，而且我不确定这是否是您想要的。或者，你可以在重复之前简单地将你的流与单个元素流连接起来，这个特殊的元素将是你的完成标记：

var completions = sequence
    .Concat(Observable.Return(42))
    .Repeat().Publish().RefCount();

如果你需要这个元素有一些上下文（你的问题没有说），把它和Observable.Defer:结合起来

var completions = sequence
    .Concat(Observable.Defer(() => Observable.Return(someExpression))
    .Repeat().Publish().RefCount();