如何实现仅当可观察序列开头匹配指定集合时才转发的Rx算子

核心逻辑 #

需要实现的Rx算子需满足以下规则：

• 若源序列IObservable<T>开头的元素与给定集合ICollection<T>的元素值、顺序完全一致，则原封不动转发整个源序列的所有元素
• 否则返回空序列
• 源序列抛出的异常会原样转发给下游观察者

边界情况处理 #

• 输入的源序列、预期集合不允许为null，否则抛出参数空异常
• 若预期集合为空，则天然匹配所有序列，直接返回原序列
• 若源序列总长度小于预期集合长度，判定为不匹配，返回空序列

可运行实现代码 #

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;

public static class ObservableExtensions
{
    public static IObservable<T> IfFirstElements<T>(
        this IObservable<T> source,
        ICollection<T> expectedFirstElements,
        IEqualityComparer<T> comparer = default)
    {
        // 参数校验
        if (source == null) throw new ArgumentNullException(nameof(source));
        if (expectedFirstElements == null) throw new ArgumentNullException(nameof(expectedFirstElements));
        comparer ??= EqualityComparer<T>.Default;

        int expectedCount = expectedFirstElements.Count;
        // 预期集合为空直接返回原序列
        if (expectedCount == 0) return source;

        var expectedList = expectedFirstElements.ToList();
        // 多播源序列避免多次订阅副作用
        return source.Publish(published =>
            // 收集前N个元素做匹配校验
            published.Take(expectedCount).ToList().SelectMany(buffer =>
            {
                // 源序列长度不足，匹配失败
                if (buffer.Count != expectedCount)
                    return Observable.Empty<T>();

                // 逐位比对元素和顺序
                for (int i = 0; i < expectedCount; i++)
                {
                    if (!comparer.Equals(buffer[i], expectedList[i]))
                        return Observable.Empty<T>();
                }

                // 匹配成功：先发送缓冲的前缀元素，再转发后续所有元素
                return buffer.ToObservable().Concat(published.Skip(expectedCount));
            })
        );
    }
}

效果验证（以预期集合[a,b,c]为例） #

1. 源序列完整输出[a,b,c,d,e]：匹配成功，输出与源序列完全一致
2. 源序列前两位为[a,p]：第二位匹配失败，输出空序列
3. 源序列仅输出[a,b]：长度不足3，匹配失败，输出空序列
4. 源序列第一个元素为[c]：第一位匹配失败，输出空序列

内容的提问来源于stack exchange，提问作者Theodor Zoulias