c# - RX Observable.TakeWhile 在每个元素之前检查条件，但我需要在之后执行检查

Question

Observable.TakeWhile 允许您在条件为真时运行序列(使用委托(delegate)以便我们可以对实际序列对象执行计算)，但它会在每个元素之前检查此条件。如何在每个元素之后执行相同的检查？

下面的代码演示了这个问题

    void RunIt()
    {
        List<SomeCommand> listOfCommands = new List<SomeCommand>();
        listOfCommands.Add(new SomeCommand { CurrentIndex = 1, TotalCount = 3 });
        listOfCommands.Add(new SomeCommand { CurrentIndex = 2, TotalCount = 3 });
        listOfCommands.Add(new SomeCommand { CurrentIndex = 3, TotalCount = 3 });

        var obs = listOfCommands.ToObservable().TakeWhile(c => c.CurrentIndex != c.TotalCount);

        obs.Subscribe(x =>
        {
            Debug.WriteLine("{0} of {1}", x.CurrentIndex, x.TotalCount);
        });
    }

    class SomeCommand
    {
        public int CurrentIndex;
        public int TotalCount;
    }

这输出

1 of 3
2 of 3

获取不到第三个元素

看到这个例子，你可能会认为我所要做的就是像这样改变我的条件 -

var obs = listOfCommands.ToObservable().TakeWhile(c => c.CurrentIndex <= c.TotalCount);

但是 observable 永远不会完成(因为在我的真实世界代码中，流不会在这三个命令之后结束)

Answer 1

没有内置运算符来执行您的要求，但这里有一个使用 Publish 运行两个查询，同时只订阅一次底层可观察对象:

// Emits matching values, but includes the value that failed the filter
public static IObservable<T> TakeWhileInclusive<T>(
    this IObservable<T> source, Func<T, bool> predicate)
{
    return source.Publish(co => co.TakeWhile(predicate)
        .Merge(co.SkipWhile(predicate).Take(1)));
}

然后:

var obs = listOfCommands.ToObservable()
    .TakeWhileInclusive(c.CurrentIndex != c.TotalCount);