Using IAsyncEnumerable in the .NET Client

[mikaelm12]

Fixes: #5376
We finally get to use the StreamAsync method name 😄

Currently what we have is

public static Task<ChannelReader<TResult>> StreamAsChannelAsync<TResult>(...)

This is adding support for

public static IAsyncEnumerable<TResult> StreamAsync<TResult>(...)

[BrennanConroy]

What if we just did

return await StreamAsChannelCoreAsync(methodName, returnType, args, cancellationToken).ReadAllAsync();

[mikaelm12]

What we should actually be doing here is using the MakeCancelableAsyncEnumerableFromChannel<T> method on the AsyncEnumerableAdapters class

[analogrelay]

Sounds good. We can just share the source for that file for now.

[Eilon]

Tagging @divega because I know you're passionate about IAsyncEnumerable, so FYI.

[analogrelay]

You shouldn't need the await foreach here anymore right? MakeCancelableAsyncEnumerable just returns the think we need. It can just be like this:

Suggested change

	await foreach(var item in AsyncEnumerableAdapters.MakeCancelableAsyncEnumerable(stream, cancellationToken)) {
	return AsyncEnumerableAdapters.MakeCancelableAsyncEnumerable(stream, cancellationToken);

[analogrelay]

Ah wait, I see why. Because that would end up being a Task<IAsyncEnumerable<T>> when we want an IAsyncEnumerable<T>. Unfortunate, but makes sense. Disregard!

[divega]

If you are going to be doing this a lot (creating IAsyncEnumerable<T> from async methods), you could consider creating a helper like MakeCancelableAsyncEnumerableFromChannel but that takes a Task<Channel<T>> and handles it to an IAsyncEnumerable<T> that defers the await until the first call to MoveNextAsync. Then you can return that directly without creating an iterator method.

The only observable difference would be that any exception that awaiting StreamAsChannelCoreAsyncCore would cause will be deferred.

This is what we do with ExecuteDataReaderAsync.

[halter73]

That's a cool idea. Probably worth making a new issue to track it. We could convert Task<ChannelReader<T>> to ChannelReader<T> in the same way.

[analogrelay]

I think that's something we can look at if we encounter an issue. Let's get this rolling now and see how it lands :).

[divega]

FWIW, one correction to what I said before: You want to pass a Func<Task<Channel<T>>> to allow deferring the call (in this case to StreamAsChannelCoreAsyncCore) completely instead of passing an already initiated Task<T>.

Probably worth making a new issue to track it.

Sure. If you see value. I am back to lurker mode 😄

[mikaelm12]

Is there a better way to do this?

[halter73]

Instead of iterating over MakeCancelableAsyncEnumerable()'s return value, just return it directly.

You might want to create another version of the method which actually uses the CancellationToken passed to GetAsyncEnumerator(), and returns IAsyncEnumerable<TResult> instead of IAsyncEnumerable<object>.

Returning a IAsyncEnumerable<TResult> is actually easier since you can always directly return the value from ChannelReader<TResult>.ReadAllAsync().GetAsyncEnumerator() from CancelableAsyncEnumerable<TResult>.GetAsyncEnumerator() without worrying about wrapping the async enumerators for value types with BoxedAsyncEnumerator.

[analogrelay]

Instead of iterating over MakeCancelableAsyncEnumerable()'s return value, just return it directly.

That doesn't really work because of the await of StreamAsChannelCoreAsyncCore beforehand though, right? That's why this has to be transformed into a generator using the yield syntax.

[halter73]

If we return the IAsyncEnumerable directly, we'd have to make StreamAsync return Task<IAsyncEnumerable<TResult>> which @anurse and I decided isn't great. I think using an iterator method is OK then.

We still should use a version of MakeCancelableAsyncEnumerable<TResult> that returns IAsyncEnumerable<TResult> and uses the CancellationToken passed through to GetAsyncEnumerator(CancellationToken) linking it with the CancellationToken from StreamAsyncCore if both are non-default.

[mikaelm12]

Task<IAsyncEnumerable> which @anurse and I decided isn't great. I think using an iterator method is OK then.

The problem with this is that if we use an iterator and yield return from the result of MakeCancelableAsyncEnumerable we actually lose the cancellation token which defeats the purpose.

We still should use a version of MakeCancelableAsyncEnumerable that returns IAsyncEnumerable

I worked down this route and realized that the streaming methods like StreamAsChannelCoreAsync all return ChannelReader<object>. This means that no matter what we would have to iterate and cast on every item.

[halter73]

Make this method generic and have it return IAsyncEnumerable<TResult> so you don't need to convert later in HubConnectionExtensions.StreamAsyncCore().

[halter73]

Move AsyncEnumerableAdapters to SignalRSharedSourceRoot if you can.

[halter73]

I would like us to create and use a new AsyncEnumerableAdapters .MakeCancelableAsyncEnumerable<TResult>() method that uses the CancellationToken from GetAsyncEnumerator().

[mikaelm12]

Changed this to return a Task<IAsyncEnumerable> to preserve the cancellation logic from the CancelableTypedAsyncEnumerable

[mikaelm12]

To be clear, I'd rather just return the IAsyncEnumerable but otherwise we lose the cancel logic

[halter73]

Can you explain this more?

Having StreamAsync return a Task<IAsyncEnumerable> is more consistent with StreamAsChannelAsync returning Task<ChannelReader<TResult>>, but in an ideal world where we had infinite development time neither would return a Task in 3.0.

Is there some reason cancellation cannot work if StreamAsync doesn't return a Task?

[mikaelm12]

Before we were doing something similar to

var stream= AsyncEnumerableAdapters.MakeCancelableTypedAsyncEnumerable(enumerable, cancellationToken)
await foreach(var item in stream)
{
    yield return stream
}

As I understand it this new AsyncEnumerable that from these returns wont have the previous cancellationToken that was passed in.

[halter73]

You can just move the call to StreamAsChannelCoreAsync into CastIAsyncEnumerable and await it at the beginning. Then StreamAsyncCore doesn't need to be async or Task-returning.

[mikaelm12]

I would like us to create and use a new AsyncEnumerableAdapters .MakeCancelableAsyncEnumerable() method that uses the CancellationToken from GetAsyncEnumerator().

We found that the implementation of IAsyncEnumerrator doesn't do anything with the cancellation token passed into GetAsyncEnumerator which is why I couldn't get that test to pass @halter73

[mikaelm12]

This is ready for review again

[halter73]

nit: maybe rename StreamAsAsyncEnumerableCore now

[mikaelm12]

@aspnet-hello triage https://dev.azure.com/dnceng/public/_build/results?buildId=145064&view=logs

[aspnet-hello]

This comment was made automatically. If there is a problem contact ryanbrandenburg.

I've triaged the above build.

[mikaelm12]

@aspnet-hello triage https://dev.azure.com/dnceng/public/_build/results?buildId=145064&view=logs

[mikaelm12]

@aspnet-hello triage https://dev.azure.com/dnceng/public/_build/results?buildId=145064&view=logs&jobId=900d802f-1272-5fca-9818-053d703b72e8