Push based coroutines

[fedeAlterio]

C# has excellent language support for pull-based streams using constructs like yield return, (await) foreach, and interfaces such as IEnumerable and IAsyncEnumerable.

However, push-based streams, represented by IObservable (and potentially IAsyncObservable) do not have this support.

Proposed Syntax: notify and (await) wheneach

IObservable<TResult> Select<TSource, TResult>(
    this IObservable<TSource> source,
    Func<TSource, TResult> selector)
{
    wheneach (var item in source)
    {
        notify selector(item);
    }
}

async IObservable<TResult> Select<TSource, TResult>(
    this IObservable<TSource> source,
    Func<TSource, Task<TResult>> selector)
{
    wheneach (var item in source)
    {
        var result = await selector(item);
        notify result;
    }
}

async IAsyncObservable<TResult> Select<TSource, TResult>(
    this IAsyncObservable<TSource> source,
    Func<TSource, TResult> selector,
    [SubscriptionCancellation] CancellationToken cancellationToken = default)
{
    await wheneach (var item in source.WithCancellation(cancellationToken))
    {
        await notify(selector(item), cancellationToken);
    }
}

Basically calling notify "teleports the execution" to the body of the wheneach

An exception should be rethrown on wheneach in an error is notified, and an exception in the coroutine should become an onError notification.

Reaching the end of the method means completing the observable.

break a wheneach means to unsubscribe from the observable

[HaloFour]

IObservable<T> has pretty robust LINQ/async support: https://github.com/dotnet/reactive

[fedeAlterio]

Yes I know :)
I think that language support can improve dev experience as much as it does for IAsyncEnumerable and IEnumerable.

In dotnet reactive if I am not wrong there was implemented an IAsyncEnumerable-like interface (and linq) before IAsyncEnumerable was a thing

[HaloFour]

Also, see: dotnet/roslyn#261