// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT License. // See the LICENSE file in the project root for more information. using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerableEx { /// /// Expands (breadth first) the async-enumerable sequence by recursively applying a selector function to generate more sequences at each recursion level. /// /// Source sequence element type. /// Source async-enumerable sequence. /// Selector function to retrieve the next sequence to expand. /// Sequence with results from the recursive expansion of the source sequence. 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 Core(source, selector); static async IAsyncEnumerable Core(IAsyncEnumerable source, Func> selector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) { var queue = new Queue>(); queue.Enqueue(source); while (queue.Count > 0) { await foreach (var item in queue.Dequeue().WithCancellation(cancellationToken).ConfigureAwait(false)) { queue.Enqueue(selector(item)); yield return item; } } } } /// /// Expands (breadth first) the async-enumerable sequence by recursively applying an asynchronous selector function to generate more sequences at each recursion level. /// /// Source sequence element type. /// Source async-enumerable sequence. /// Asynchronous selector function to retrieve the next sequence to expand. /// Sequence with results from the recursive expansion of the source sequence. 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 Core(source, selector); static async IAsyncEnumerable Core(IAsyncEnumerable source, Func>> selector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) { var queue = new Queue>(); queue.Enqueue(source); while (queue.Count > 0) { await foreach (var item in queue.Dequeue().WithCancellation(cancellationToken).ConfigureAwait(false)) { queue.Enqueue(await selector(item).ConfigureAwait(false)); yield return item; } } } } #if !NO_DEEP_CANCELLATION /// /// Expands (breadth first) the async-enumerable sequence by recursively applying an asynchronous (cancellable) selector function to generate more sequences at each recursion level. /// /// Source sequence element type. /// Source async-enumerable sequence. /// Asynchronous (cancellable) selector function to retrieve the next sequence to expand. /// Sequence with results from the recursive expansion of the source sequence. 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 Core(source, selector); static async IAsyncEnumerable Core(IAsyncEnumerable source, Func>> selector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) { var queue = new Queue>(); queue.Enqueue(source); while (queue.Count > 0) { await foreach (var item in queue.Dequeue().WithCancellation(cancellationToken).ConfigureAwait(false)) { queue.Enqueue(await selector(item, cancellationToken).ConfigureAwait(false)); yield return item; } } } } #endif } }