Reactive: Better support for async - Task based overloads for AsyncEnumerable

[clairernovotny]

As mentioned in several Issues and PR's: #96, #170, #168, #144 and other threads.

We need to add async overloads to all of the query operators in AsyncEnumerable to support async callbacks, accumulators and the rest.

From an API design perspective, do we want to add overloads or create new *Async versions of the methods? Do the language integrated query keywords work with async such that we _need_ to maintain the current naming to work there?

[clairernovotny]

We should probably wait until preview 4 of VS "15" for this based on @ljw1004's thoughts:
http://www.twitter.com/lwischik/status/755933011724476416

Seems like this will help perf here as it'll be an inner-loop. As ValueTask is in a public package, we can start exposing this before ppl can easily use it and once people use C# 7, it'll "just work."

[ljw1004]

Agreed. If you're operating over a sequence, you should expose something like

IAsyncEnumerable<U> SelectAsync<T,U>(this IEnumerable<T> seq, Func<T,ValueTask<U>> lambda)

with ValueTask not Task.

The reason is that presumably this lambda will be invoked once for each element in the sequence. It's typical for folks to write an async lambda which returns synchronously 99.9% of the time, and only shells out to async on the rare occasions that it needs to pull in more data. Using ValueTask will avoid one heap allocation per element on those 99.9% of times.

You should not offer two overloads

IAsyncEnumerable<U> SelectAsync<T,U>(this IEnumerable<T> seq, Func<T,ValueTask<U>> lambda)
IAsyncEnumerable<U> SelectAsync<T,U>(this IEnumerable<T> seq, Func<T,Task<U>> lambda)

If you did, then someone who invokes seq.SelectAsync(async (x) => {...; return x+1;}) in C#6 would get the Task overload, and in C#7 would get an ambiguity error.

[clairernovotny]

@ljw1004 You bring up a good / different point about the surface area that we hadn't considered but probably should - async accumulators to IEnumerable that turn into IAsyncEnumerable.

Originally, I was thinking of

IAsyncEnumerable<U> Select<T,U>(this IAsyncEnumerable<T> seq, Func<T,ValueTask<U>> lambda)

But your use-case is an equally valid way to get an IAsyncEnumerable sequence from a "regular" IEnumerable

BTW, in this one case, we cannot use the *Async suffix because of the LINQ keyword -> method expansion pattern. We have to use Select() for thing to work there as expected.

[ljw1004]

BTW, in this one case, we cannot use the *Async suffix because of the LINQ keyword -> method expansion pattern. We have to use Select() for thing to work there as expected.

Technically, there is no keyword->method expansion pattern yet for LINQ expressions that have "await" in them. So once you want to support that, I think you should consider it up for discussion.

If you don't adopt the *Async suffix, I have a hunch there'll be ambiguity... For instance,

IAsyncEnumerable<U> Select<T,U>(this IEnumerable<T> seq, Func<T, ValueTask<U>> lambda)
IEnumerable<U> Select<T,U>(this IEnumerable<T> seq, Func<T,U> lambda)

Select(src, (x) => Task.Delay(x)); // picks the second overload with U=Task
Select(src, (x) => ValueTaskReturningMethod()); // picks the first overload with U=int
Select(src, async (x) => x+1); // in C#6 picks the second, but in C#7 picks the first

I worry that it might get subtle to have the return type differ so dramatically for sigh slight reasons.

[clairernovotny]

We already do support LINQ expressions on IAsyncEnumerable though today since we don't have async lambda's.

IAsyncEnumerable<T> source = ...;

IAsyncEnumerable<U> result = from item in source
                             where item.Foo == bar
                             select item.Bar;

We get that because the existing methods are implemented as Select and Where, etc. If/when LINQ syntax is mixed in, we'd want to have the Func<T, ValueTask<U>> version just fit in.

Or are you suggesting that we explicitly make the Func<T, ValueTask<U>> versions as SelectAsync because there's no current support within LINQ and some future expansion will support that? It'd be nice to be on the same page as the languages team here so we don't get cross-wired :)

[i3arnon]

Are we still waiting? (for async streams I guess, as we already got task-like returning async methods)
Or can we start going in the direction of supporting async delegates?
If so I would happily start contributing them.

[bradphelan]

Does anybody have a prototype impl of what SelectAsync would look like so I can polyfill this till it makes it way into the library?

[bradphelan]

This seems to do the job

    public static class AsyncEnumerableExtensions
    {
        public class SelectAsyncEnumerator<T,U> : IAsyncEnumerator<U>
        {
            private readonly IAsyncEnumerator<T> _Source;
            private readonly Func<T, Task<U>> _Func;

            public SelectAsyncEnumerator(IAsyncEnumerator<T> source, Func<T, Task<U>> func)
            {
                _Source = source;
                _Func = func;
            }

            public void Dispose()
            {
                _Source.Dispose();
            }

            public async Task<bool> MoveNext(CancellationToken cancellationToken)
            {
                if (!await _Source.MoveNext())
                    return false;
                Current = await _Func( _Source.Current );
                return true;
            }

            public U Current { get; private set; }
        }

        public class SelectAsyncEnumerable<T, U> : IAsyncEnumerable<U>
        {
            private readonly IAsyncEnumerable<T> _Source;
            private readonly Func<T, Task<U>> _Func;

            public SelectAsyncEnumerable(IAsyncEnumerable<T> source, Func<T, Task<U>> func)
            {
                _Source = source;
                _Func = func;
            }

            public IAsyncEnumerator<U> GetEnumerator() => new SelectAsyncEnumerator<T,U>(_Source.GetEnumerator(), _Func);
        }

        public static IAsyncEnumerable<U> SelectAsync<T, U>(this IAsyncEnumerable<T> source, Func<T, Task<U>> fn)
        {
            return new SelectAsyncEnumerable<T,U>( source, fn );
        }
    }

[clairernovotny]

These are now available in the 4.0 preview