// 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
}
}