// 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.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { public static Task AggregateAsync(this IAsyncEnumerable source, Func accumulator, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); return Core(source, accumulator, cancellationToken); static async Task Core(IAsyncEnumerable _source, Func _accumulator, CancellationToken _cancellationToken) { #if USE_AWAIT_USING await using (var e = _source.GetAsyncEnumerator(_cancellationToken).ConfigureAwait(false)) { if (!await e.MoveNextAsync()) { throw Error.NoElements(); } var acc = e.Current; while (await e.MoveNextAsync()) { acc = _accumulator(acc, e.Current); } return acc; } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { if (!await e.MoveNextAsync().ConfigureAwait(false)) { throw Error.NoElements(); } var acc = e.Current; while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = _accumulator(acc, e.Current); } return acc; } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif } } public static Task AggregateAsync(this IAsyncEnumerable source, Func> accumulator, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); return Core(source, accumulator, cancellationToken); static async Task Core(IAsyncEnumerable _source, Func> _accumulator, CancellationToken _cancellationToken) { #if USE_AWAIT_USING await using (var e = _source.GetAsyncEnumerator(_cancellationToken).ConfigureAwait(false)) { if (!await e.MoveNextAsync()) { throw Error.NoElements(); } var acc = e.Current; while (await e.MoveNextAsync()) { acc = await _accumulator(acc, e.Current).ConfigureAwait(false); } return acc; } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { if (!await e.MoveNextAsync().ConfigureAwait(false)) { throw Error.NoElements(); } var acc = e.Current; while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = await _accumulator(acc, e.Current).ConfigureAwait(false); } return acc; } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif } } #if !NO_DEEP_CANCELLATION public static Task AggregateAsync(this IAsyncEnumerable source, Func> accumulator, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); return Core(source, accumulator, cancellationToken); static async Task Core(IAsyncEnumerable _source, Func> _accumulator, CancellationToken _cancellationToken) { #if USE_AWAIT_USING await using (var e = _source.GetAsyncEnumerator(_cancellationToken).ConfigureAwait(false)) { if (!await e.MoveNextAsync()) { throw Error.NoElements(); } var acc = e.Current; while (await e.MoveNextAsync()) { acc = await _accumulator(acc, e.Current, _cancellationToken).ConfigureAwait(false); } return acc; } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { if (!await e.MoveNextAsync().ConfigureAwait(false)) { throw Error.NoElements(); } var acc = e.Current; while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = await _accumulator(acc, e.Current, _cancellationToken).ConfigureAwait(false); } return acc; } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif } } #endif public static Task AggregateAsync(this IAsyncEnumerable source, TAccumulate seed, Func accumulator, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); return Core(source, seed, accumulator, cancellationToken); static async Task Core(IAsyncEnumerable _source, TAccumulate _seed, Func _accumulator, CancellationToken _cancellationToken) { var acc = _seed; #if USE_AWAIT_FOREACH await foreach (TSource item in _source.WithCancellation(_cancellationToken).ConfigureAwait(false)) { acc = _accumulator(acc, item); } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = _accumulator(acc, e.Current); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return acc; } } public static Task AggregateAsync(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); return Core(source, seed, accumulator, cancellationToken); static async Task Core(IAsyncEnumerable _source, TAccumulate _seed, Func> _accumulator, CancellationToken _cancellationToken) { var acc = _seed; #if USE_AWAIT_FOREACH await foreach (TSource item in _source.WithCancellation(_cancellationToken).ConfigureAwait(false)) { acc = await _accumulator(acc, item).ConfigureAwait(false); } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = await _accumulator(acc, e.Current).ConfigureAwait(false); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return acc; } } #if !NO_DEEP_CANCELLATION public static Task AggregateAsync(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); return Core(source, seed, accumulator, cancellationToken); static async Task Core(IAsyncEnumerable _source, TAccumulate _seed, Func> _accumulator, CancellationToken _cancellationToken) { var acc = _seed; #if USE_AWAIT_FOREACH await foreach (TSource item in _source.WithCancellation(_cancellationToken).ConfigureAwait(false)) { acc = await _accumulator(acc, item, _cancellationToken).ConfigureAwait(false); } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = await _accumulator(acc, e.Current, _cancellationToken).ConfigureAwait(false); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return acc; } } #endif public static Task AggregateAsync(this IAsyncEnumerable source, TAccumulate seed, Func accumulator, Func resultSelector, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); return Core(source, seed, accumulator, resultSelector, cancellationToken); static async Task Core(IAsyncEnumerable _source, TAccumulate _seed, Func _accumulator, Func _resultSelector, CancellationToken _cancellationToken) { var acc = _seed; #if USE_AWAIT_FOREACH await foreach (TSource item in _source.WithCancellation(_cancellationToken).ConfigureAwait(false)) { acc = _accumulator(acc, item); } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = _accumulator(acc, e.Current); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return _resultSelector(acc); } } public static Task AggregateAsync(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, Func> resultSelector, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); return Core(source, seed, accumulator, resultSelector, cancellationToken); static async Task Core(IAsyncEnumerable _source, TAccumulate _seed, Func> _accumulator, Func> _resultSelector, CancellationToken _cancellationToken) { var acc = _seed; #if USE_AWAIT_FOREACH await foreach (TSource item in _source.WithCancellation(_cancellationToken).ConfigureAwait(false)) { acc = await _accumulator(acc, item).ConfigureAwait(false); } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = await _accumulator(acc, e.Current).ConfigureAwait(false); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return await _resultSelector(acc).ConfigureAwait(false); } } #if !NO_DEEP_CANCELLATION public static Task AggregateAsync(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, Func> resultSelector, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (accumulator == null) throw Error.ArgumentNull(nameof(accumulator)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); return Core(source, seed, accumulator, resultSelector, cancellationToken); static async Task Core(IAsyncEnumerable _source, TAccumulate _seed, Func> _accumulator, Func> _resultSelector, CancellationToken _cancellationToken) { var acc = _seed; #if USE_AWAIT_FOREACH await foreach (TSource item in _source.WithCancellation(_cancellationToken).ConfigureAwait(false)) { acc = await _accumulator(acc, item, _cancellationToken).ConfigureAwait(false); } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { acc = await _accumulator(acc, e.Current, _cancellationToken).ConfigureAwait(false); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return await _resultSelector(acc, _cancellationToken).ConfigureAwait(false); } } #endif } }