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

Question

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

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

Answer 1

没有内置运算符可以满足您的要求，但这里有一个使用 Publish 运行两个查询，同时只订阅底层 observable 一次：

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

Answer 2

最终编辑：

我的解决方案基于 Sergey 在此线程中的 TakeWhileInclusive 实现 - How to complete a Rx Observable depending on a condition in a event

public static IObservable<TSource> TakeUntil<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
    ));
}

Answer 3

您可以使用TakeUntil 运算符来获取每个项目，直到第二个来源产生一个值；在这种情况下，我们可以将第二个流指定为谓词通过后的第一个值：

public static IObservable<TSource> TakeWhileInclusive<TSource>(
    this IObservable<TSource> source,
    Func<TSource, bool> predicate)
{
    return source.TakeUntil(source.SkipWhile(x => predicate(x)).Skip(1));
}

Answer 4

我认为你在关注TakeWhile，而不是TakeUntil：

var list = (new List<int>(){1,2,3,4,5,6,7,8,9,10});
var takeWhile = list
        .ToObservable()
        .Select((_, i) => Tuple.Create(i, _))
        .TakeWhile(tup => tup.Item1 < list.Count)
        .Do(_ => Console.WriteLine("Outputting {0}", _.Item2));

好的，您想要的东西不存在开箱即用，至少我不知道具有这种特定语法的东西。也就是说，你可以很容易地把它拼凑起来（而且它不是太讨厌）：

var fakeCmds = Enumerable
    .Range(1, 100)
    .Select(i => new SomeCommand() {CurrentIndex = i, TotalCount = 10})
    .ToObservable();

var beforeMatch = fakeCmds
    .TakeWhile(c => c.CurrentIndex != c.TotalCount);
var theMatch = fakeCmds
    .SkipWhile(c => c.CurrentIndex != c.TotalCount)
    .TakeWhile(c => c.CurrentIndex == c.TotalCount);
var upToAndIncluding = Observable.Concat(beforeMatch, theMatch);

Answer 5

组合，使用新的SkipUntil 和TakeUntil：

SkipUntil return source.Publish(s => s.SkipUntil(s.Where(predicate)));

TakeUntil （含） return source.Publish(s => s.TakeUntil(s.SkipUntil(predicate)));

完整来源：https://gist.github.com/GeorgeTsiokos/a4985b812c4048c428a981468a965a86

Answer 6

也许以下方式对某人有用。 您必须使用“Do”方法和空的“Subscribe”方法。

    listOfCommands.ToObservable()
    .Do(x =>
    {
        Debug.WriteLine("{0} of {1}", x.CurrentIndex, x.TotalCount);
    })
    .TakeWhile(c => c.CurrentIndex != c.TotalCount)
    .Subscribe();

这样您无需编写自己的扩展即可获得结果。