cleaned up IAsyncEnumerable Source to use local functions

[Aaronontheweb]

Changes

Better to use await around ValueTask rather than converting to Task and doing ContinueWith

Cleanup from #6044

Checklist

For significant changes, please ensure that the following have been completed (delete if not relevant):

• This change follows the Akka.NET API Compatibility Guidelines.
• This change follows the Akka.NET Wire Compatibility Guidelines.
• I have reviewed my own pull request.

[Arkatufus]

Looks like the code can be cleaned up further.

[Arkatufus]

@Aaronontheweb Need you to review the changes I've made

[Aaronontheweb]

LGTM

[to11mtm]

Interesting question, should this be in prestart vs the constructor?

[Aaronontheweb]

The only thing that would throw, I guess, would be if there was something wrong with the IAsyncEnumerable (i.e. null)?

In that case I guess it'd be better to have this run in PreStart

[Arkatufus]

That's a good point, I'll make the appropriate changes

[to11mtm]

Do we have guarantees that cancelling will dispose the enumerator if needed?

[Arkatufus]

Oh, you're totally correct, I forgot that we need to dispose the enumerator... I guess the previous code is still needed here.

[Arkatufus]

Here's a big question, should we defer the base.OnDownstreamFinish() call and do this asynchronously or just block and wait until _enumerator.DisposeAsync is complete?

[Arkatufus]

            public override void OnDownstreamFinish(Exception cause)
            {
                _completionCts.Cancel();
                _completionCts.Dispose();
                _enumerator.DisposeAsync().AsTask().ContinueWith(t => _baseOnDownstreamFinishCallback(cause));
            }

            private void BaseOnDownstreamFinishCallback(Exception cause)
            {
                CompleteStage();
                base.OnDownstreamFinish(cause);
            }

or

            public override void OnDownstreamFinish(Exception cause)
            {
                _completionCts.Cancel();
                _completionCts.Dispose();
                CompleteStage();

                _enumerator.DisposeAsync().GetAwaiter().GetResult();
                base.OnDownstreamFinish(cause);
            }

[Arkatufus]

I will add the try...catch later, just wanted to get the point across first

[Arkatufus]

The only problem that I can see here is a custom IAsyncEnumerator that deadlocked/froze when disposed

[Aaronontheweb]

I think we should go with the latter - I doubt there will be IAsyncEnumerator.DisposeAsync implementations that can block for prolonged periods of time.

[Arkatufus]

Done

[Aaronontheweb]

LGTM