// 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
{
// NB: Implementations of Scan never yield the first element, unlike the behavior of Aggregate on a sequence with one
// element, which returns the first element (or the seed if given an empty sequence). This is compatible with Rx
// but one could argue whether it was the right default.
///
/// Applies an accumulator function over an async-enumerable sequence and returns each intermediate result.
/// For aggregation behavior with no intermediate results, see .
///
/// The type of the elements in the source sequence and the result of the aggregation.
/// An async-enumerable sequence to accumulate over.
/// An accumulator function to be invoked on each element.
/// An async-enumerable sequence containing the accumulated values.
/// or is null.
public static IAsyncEnumerable Scan(this IAsyncEnumerable source, Func accumulator)
{
if (source == null)
throw Error.ArgumentNull(nameof(source));
if (accumulator == null)
throw Error.ArgumentNull(nameof(accumulator));
return Core(source, accumulator);
static async IAsyncEnumerable Core(IAsyncEnumerable source, Func accumulator, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
await using var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false);
if (!await e.MoveNextAsync())
{
yield break;
}
var res = e.Current;
while (await e.MoveNextAsync())
{
res = accumulator(res, e.Current);
yield return res;
}
}
}
///
/// Applies an accumulator function over an async-enumerable sequence and returns each intermediate result. The specified seed value is used as the initial accumulator value.
/// For aggregation behavior with no intermediate results, see .
///
/// The type of the elements in the source sequence.
/// The type of the result of the aggregation.
/// An async-enumerable sequence to accumulate over.
/// The initial accumulator value.
/// An accumulator function to be invoked on each element.
/// An async-enumerable sequence containing the accumulated values.
/// or is null.
public static IAsyncEnumerable Scan(this IAsyncEnumerable source, TAccumulate seed, Func accumulator)
{
if (source == null)
throw Error.ArgumentNull(nameof(source));
if (accumulator == null)
throw Error.ArgumentNull(nameof(accumulator));
return Core(source, seed, accumulator);
static async IAsyncEnumerable Core(IAsyncEnumerable source, TAccumulate seed, Func accumulator, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
var res = seed;
await foreach (var item in source.WithCancellation(cancellationToken).ConfigureAwait(false))
{
res = accumulator(res, item);
yield return res;
}
}
}
///
/// Applies an asynchronous accumulator function over an async-enumerable sequence and returns each intermediate result.
/// For aggregation behavior with no intermediate results, see .
///
/// The type of the elements in the source sequence and the result of the aggregation.
/// An async-enumerable sequence to accumulate over.
/// An asynchronous accumulator function to be invoked and awaited on each element.
/// An async-enumerable sequence containing the accumulated values.
/// or is null.
public static IAsyncEnumerable Scan(this IAsyncEnumerable source, Func> accumulator)
{
if (source == null)
throw Error.ArgumentNull(nameof(source));
if (accumulator == null)
throw Error.ArgumentNull(nameof(accumulator));
return Core(source, accumulator);
static async IAsyncEnumerable Core(IAsyncEnumerable source, Func> accumulator, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
await using var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false);
if (!await e.MoveNextAsync())
{
yield break;
}
var res = e.Current;
while (await e.MoveNextAsync())
{
res = await accumulator(res, e.Current).ConfigureAwait(false);
yield return res;
}
}
}
#if !NO_DEEP_CANCELLATION
///
/// Applies an asynchronous (cancellable) accumulator function over an async-enumerable sequence and returns each intermediate result.
/// For aggregation behavior with no intermediate results, see .
///
/// The type of the elements in the source sequence and the result of the aggregation.
/// An async-enumerable sequence to accumulate over.
/// An asynchronous (cancellable) accumulator function to be invoked and awaited on each element.
/// An async-enumerable sequence containing the accumulated values.
/// or is null.
public static IAsyncEnumerable Scan(this IAsyncEnumerable source, Func> accumulator)
{
if (source == null)
throw Error.ArgumentNull(nameof(source));
if (accumulator == null)
throw Error.ArgumentNull(nameof(accumulator));
return Core(source, accumulator);
static async IAsyncEnumerable Core(IAsyncEnumerable source, Func> accumulator, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
await using var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false);
if (!await e.MoveNextAsync())
{
yield break;
}
var res = e.Current;
while (await e.MoveNextAsync())
{
res = await accumulator(res, e.Current, cancellationToken).ConfigureAwait(false);
yield return res;
}
}
}
#endif
///
/// Applies an asynchronous accumulator function over an async-enumerable sequence and returns each intermediate result. The specified seed value is used as the initial accumulator value.
/// For aggregation behavior with no intermediate results, see .
///
/// The type of the elements in the source sequence.
/// The type of the result of the aggregation.
/// An async-enumerable sequence to accumulate over.
/// The initial accumulator value.
/// An asynchronous accumulator function to be invoked on each element.
/// An async-enumerable sequence containing the accumulated values.
/// or is null.
public static IAsyncEnumerable Scan(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator)
{
if (source == null)
throw Error.ArgumentNull(nameof(source));
if (accumulator == null)
throw Error.ArgumentNull(nameof(accumulator));
return Core(source, seed, accumulator);
static async IAsyncEnumerable Core(IAsyncEnumerable source, TAccumulate seed, Func> accumulator, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
var res = seed;
await foreach (var item in source.WithCancellation(cancellationToken).ConfigureAwait(false))
{
res = await accumulator(res, item).ConfigureAwait(false);
yield return res;
}
}
}
#if !NO_DEEP_CANCELLATION
///
/// Applies an asynchronous (cancellable) accumulator function over an async-enumerable sequence and returns each intermediate result. The specified seed value is used as the initial accumulator value.
/// For aggregation behavior with no intermediate results, see .
///
/// The type of the elements in the source sequence.
/// The type of the result of the aggregation.
/// An async-enumerable sequence to accumulate over.
/// The initial accumulator value.
/// An asynchronous (cancellable) accumulator function to be invoked on each element.
/// An async-enumerable sequence containing the accumulated values.
/// or is null.
public static IAsyncEnumerable Scan(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator)
{
if (source == null)
throw Error.ArgumentNull(nameof(source));
if (accumulator == null)
throw Error.ArgumentNull(nameof(accumulator));
return Core(source, seed, accumulator);
static async IAsyncEnumerable Core(IAsyncEnumerable source, TAccumulate seed, Func> accumulator, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
var res = seed;
await foreach (var item in source.WithCancellation(cancellationToken).ConfigureAwait(false))
{
res = await accumulator(res, item, cancellationToken).ConfigureAwait(false);
yield return res;
}
}
}
#endif
}
}