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