为什么可以等待一个 Rx observable？ [复制]答案

【问题标题】：Why is it possible to await an Rx observable? [duplicate]为什么可以等待一个 Rx observable？ [复制]
【发布时间】：2014-12-11 03:53:35
【问题描述】：

我刚刚注意到await关键字可以和Rx Observable一起使用，例如：

await Observable.Interval(TimeSpan.FromHours(1));

我很确定它只能与任务结合使用。

那么是什么让它成为可能呢？可观察对象的知识是否硬编码到编译器中？

【问题讨论】：

标签： c# async-await system.reactive

【解决方案1】：

不，编译器没有IObservable<T>的特殊知识。根据 C# 5 规范的第 7.7.7.1 节，如果对象有一个方法，或者在名为 GetAwaiter 的范围内有一个扩展方法，它返回一个实现 System.Runtime.CompilerServices.INotifyCompletion 的类型，则可以等待它。请参阅 Steven Toub 的文章，Await anything。

更具体地说，来自规范

await 表达式的任务必须是可等待的。如果满足以下条件之一，则表达式 t 是 awaitable：
- t 是动态编译时类型
- t 有一个名为 GetAwaiter 的可访问实例或扩展方法，没有参数也没有类型参数，并且返回类型 A 满足以下所有条件：
1、A实现接口System.Runtime.CompilerServices.INotifyCompletion（以下简称INotifyCompletion）
2. A 有一个可访问、可读的实例属性 IsCompleted，类型为 bool
3.A有一个可访问的实例方法GetResult，没有参数也没有类型参数

请注意，这与 foreach 不需要 IEnumerable<T> 而只是一个返回兼容对象的 GetEnumerator 方法相似。这种鸭子类型是一种性能优化，它允许编译器使用值类型而无需装箱。这可用于避免在性能敏感代码中进行不必要的分配。

【讨论】：

可能值得注意的是，等待可观察对象仅在调用 OnCompleted 时返回最后一个值。如果序列中没有任何值，则会抛出异常。

【解决方案2】：

我认为这是因为System.Reactive.Linq 在IObservable 上定义了一个GetAwaiter 扩展方法。正如@mike z 解释的那样，这允许您等待IObservable。方法如下：

public static AsyncSubject<TSource> GetAwaiter<TSource>(this IObservable<TSource> source)
{
    if (source == null)
    {
        throw new ArgumentNullException("source");
    }
    return s_impl.GetAwaiter<TSource>(source);
}

返回的类型AsyncSubject<T>实现INotifyCompletion并具有IsCompleted属性和GetResult方法。

public sealed class AsyncSubject<T> : ISubject<T>, ISubject<T, T>, IObserver<T>, IObservable<T>, IDisposable, INotifyCompletion
{
    // Fields
    private Exception _exception;
    private readonly object _gate;
    private bool _hasValue;
    private bool _isDisposed;
    private bool _isStopped;
    private ImmutableList<IObserver<T>> _observers;
    private T _value;

    // Methods
    public AsyncSubject();
    private void CheckDisposed();
    public void Dispose();
    public AsyncSubject<T> GetAwaiter();
    public T GetResult();
    public void OnCompleted();
    public void OnCompleted(Action continuation);
    private void OnCompleted(Action continuation, bool originalContext);
    public void OnError(Exception error);
    public void OnNext(T value);
    public IDisposable Subscribe(IObserver<T> observer);

    // Properties
    public bool HasObservers { get; }
    public bool IsCompleted { get; }

【讨论】：