c#-4.0 - IObservable 的 Where() 扩展方法是如何实现的？

Question

我一直在尝试更深入地了解 Rx，通过关注 Bart De Smetts MinLinq 和 Jon Skeets 的“重新实现”系列，我已经建立了很好的理解，但是......

以下面的代码为例

var onePerSecond = Observable.Interval(TimeSpan.FromSeconds(1));
var evenNums = onePerSecond.Where(x => x % 2 == 0);
evenNums.Subscribe(Console.WriteLine);

从等效的 IEnumerable 的角度来看，我了解 MoveNext/Current 的数据流，并从 Skeets 先生的博客中了解如何通过在扩展方法的 IEnumerable 'this' 参数上使用 foreach 来实现 Where 方法。

但是在 IObservable 的 Where 方法的情况下，它是否包含实现 IObserver 接口(interface)(或 lambda 等效项)的代码，因此有效地观察来自 onePerSecond 对象的所有通知，并反过来返回仅包含谓词找到的值的 IObservable是真实的？

非常欢迎任何帮助和想法，非常感谢

詹姆士

Answer 1

通过使用 ILSpy 查看源代码，很容易看出这正是 Where 的实现方式。它返回一个新的 observable，它根据您传递的谓词过滤项目:

public static IObservable<TSource> Where<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
{
    if (source == null)
    {
        throw new ArgumentNullException("source");
    }
    if (predicate == null)
    {
        throw new ArgumentNullException("predicate");
    }
    return new AnonymousObservable<TSource>((IObserver<TSource> observer) => source.Subscribe(delegate(TSource x)
    {
        bool flag;
        try
        {
            flag = predicate(x);
        }
        catch (Exception error)
        {
            observer.OnError(error);
            return;
        }
        if (flag)
        {
            observer.OnNext(x);
        }
    }
    , new Action<Exception>(observer.OnError), new Action(observer.OnCompleted)));
}