Why is Rx observation expected?

Question

I just noticed that the await keyword can be used with Rx Observable, for example:

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

I was sure that it can only be used in conjunction with tasks.

So what does it give? Is knowledge of observables hardcoded in the compiler?

Answer 1

No, the compiler does not have special knowledge of IObservable<T> . According to section 7.7.7.1 of the C # 5 specification, if an object has a method or an extension method exists in an area called GetAwaiter that returns a type that implements System.Runtime.CompilerServices.INotifyCompletion , it can be expected. See Stephen Tube's article, Expect Something .

In particular, from the specification

The task of expressing expectations should be expected. The expression t is expected if one of the following conditions is true:
- t - dynamic type of compilation time
- t has an available instance or extension method called GetAwaiter with no parameters and no type parameters, and return type A, for which all of the following operations are performed:
1. A implements the System.Runtime.CompilerServices.INotifyCompletion interface (hereinafter referred to as INotifyCompletion for short)
2. A has an accessible, readable property of an IsCompleted instance of type bool
3. A has an available GetResult instance method without parameters and type parameters

Notice how it looks like foreach does not require an IEnumerable<T> , but just a GetEnumerator method that returns a compatible object. This type of duck print is a performance optimization that allows the compiler to use value types without a box. This can be used to avoid unnecessary allocations in performance-sensitive code.

Answer 2

I think this is because System.Reactive.Linq defines the GetAwaiter extension GetAwaiter on IObservable . As @mike z explained, you can wait for IObservable . Here is a way:

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

The return type AsyncSubject<T> implements INotifyCompletion and has the IsCompleted and 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; }