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

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

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

    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);

但是可观察的将永远不会完成（因为在我的现实世界代码中，流在这三个命令之后不会结束）

没有内置的运算符可以完成您所要求的操作，但这里有一个使用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);