// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information. using System.Collections.Generic; using System.Diagnostics; using System.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerableEx { public static IAsyncEnumerable Expand(this IAsyncEnumerable source, Func> selector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (selector == null) throw Error.ArgumentNull(nameof(selector)); return new ExpandAsyncIterator(source, selector); } public static IAsyncEnumerable Expand(this IAsyncEnumerable source, Func>> selector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (selector == null) throw Error.ArgumentNull(nameof(selector)); return new ExpandAsyncIteratorWithTask(source, selector); } #if !NO_DEEP_CANCELLATION public static IAsyncEnumerable Expand(this IAsyncEnumerable source, Func>> selector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (selector == null) throw Error.ArgumentNull(nameof(selector)); return new ExpandAsyncIteratorWithTaskAndCancellation(source, selector); } #endif private sealed class ExpandAsyncIterator : AsyncIterator { private readonly Func> _selector; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private Queue> _queue; public ExpandAsyncIterator(IAsyncEnumerable source, Func> selector) { Debug.Assert(source != null); Debug.Assert(selector != null); _source = source; _selector = selector; } public override AsyncIteratorBase Clone() { return new ExpandAsyncIterator(_source, _selector); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } _queue = null; await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _queue = new Queue>(); _queue.Enqueue(_source); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (true) { if (_enumerator == null) { if (_queue.Count > 0) { var src = _queue.Dequeue(); if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); } _enumerator = src.GetAsyncEnumerator(_cancellationToken); continue; // loop } break; // while } if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; var next = _selector(item); _queue.Enqueue(next); _current = item; return true; } await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } private sealed class ExpandAsyncIteratorWithTask : AsyncIterator { private readonly Func>> _selector; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private Queue> _queue; public ExpandAsyncIteratorWithTask(IAsyncEnumerable source, Func>> selector) { Debug.Assert(source != null); Debug.Assert(selector != null); _source = source; _selector = selector; } public override AsyncIteratorBase Clone() { return new ExpandAsyncIteratorWithTask(_source, _selector); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } _queue = null; await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _queue = new Queue>(); _queue.Enqueue(_source); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (true) { if (_enumerator == null) { if (_queue.Count > 0) { var src = _queue.Dequeue(); if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); } _enumerator = src.GetAsyncEnumerator(_cancellationToken); continue; // loop } break; // while } if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; var next = await _selector(item).ConfigureAwait(false); _queue.Enqueue(next); _current = item; return true; } await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } #if !NO_DEEP_CANCELLATION private sealed class ExpandAsyncIteratorWithTaskAndCancellation : AsyncIterator { private readonly Func>> _selector; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private Queue> _queue; public ExpandAsyncIteratorWithTaskAndCancellation(IAsyncEnumerable source, Func>> selector) { Debug.Assert(source != null); Debug.Assert(selector != null); _source = source; _selector = selector; } public override AsyncIteratorBase Clone() { return new ExpandAsyncIteratorWithTaskAndCancellation(_source, _selector); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } _queue = null; await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _queue = new Queue>(); _queue.Enqueue(_source); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (true) { if (_enumerator == null) { if (_queue.Count > 0) { var src = _queue.Dequeue(); if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); } _enumerator = src.GetAsyncEnumerator(_cancellationToken); continue; // loop } break; // while } if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; var next = await _selector(item, _cancellationToken).ConfigureAwait(false); _queue.Enqueue(next); _current = item; return true; } await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } #endif } }