reactive-programming - 是否有像 groupBy 这样的 Rx 运算符只对连续的元素进行分组？

Question

我想要一个 GroupBy，它在每次键更改时发出一个新的可观察对象。有这种事吗？

像这样的伪代码:

[1,3,5,2,4,3,7] groupByConsecutive? oddEven

会产生

Observable[Observable[1,3,5],Observable[2,4],Observable[3,7]]

Answer 1

我花了一段时间想出一些不依赖于 Subject 的东西，但我认为这会做你想做的事:

new int[] { 1, 3, 5, 2, 4, 3, 7 }
.ToObservable()
.Publish(let =>
    let
    .GroupByUntil(
        x => x % 2,
        group => let.SkipWhile(x => x % 2 == group.Key)))

您可以将上面的代码重构为更通用的扩展方法，如下所示:

public static class EnumerableExtensions
{
    public static IObservable<IGroupedObservable<TKey, TSource>> GroupByUntilChanged<TSource, TKey>(
        this IObservable<TSource> source,
        Func<TSource, TKey> keySelector,
        IEqualityComparer<TKey> comparer = null)
    {
        comparer = comparer ?? EqualityComparer<TKey>.Default;

        return 
            source
            .Publish(let =>
                let
                .GroupByUntil(
                    keySelector,
                    group => let.SkipWhile(x => comparer.Equals(keySelector(x), group.Key))));
    }
}

然后像这样调用它:

new int[] { 1, 3, 5, 2, 4, 3, 7 }
.ToObservable()
.GroupByUntilChanged(x => x % 2)