// 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 Aggregate(this IAsyncEnumerable source, Func accumulator) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return Aggregate(source, accumulator, CancellationToken.None); } public static Task Aggregate(this IAsyncEnumerable source, Func accumulator, CancellationToken cancellationToken) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return Aggregate_(source, accumulator, cancellationToken); } public static Task Aggregate(this IAsyncEnumerable source, Func> accumulator) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return Aggregate(source, accumulator, CancellationToken.None); } public static Task Aggregate(this IAsyncEnumerable source, Func> accumulator, CancellationToken cancellationToken) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return Aggregate_(source, accumulator, cancellationToken); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func accumulator) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return Aggregate(source, seed, accumulator, CancellationToken.None); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func accumulator, CancellationToken cancellationToken) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return source.Aggregate(seed, accumulator, x => x, cancellationToken); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return Aggregate(source, seed, accumulator, CancellationToken.None); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, CancellationToken cancellationToken) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); return source.Aggregate(seed, accumulator, x => Task.FromResult(x), cancellationToken); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func accumulator, Func resultSelector) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); if (resultSelector == null) throw new ArgumentNullException(nameof(resultSelector)); return Aggregate(source, seed, accumulator, resultSelector, CancellationToken.None); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func accumulator, Func resultSelector, CancellationToken cancellationToken) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); if (resultSelector == null) throw new ArgumentNullException(nameof(resultSelector)); return Aggregate_(source, seed, accumulator, resultSelector, cancellationToken); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, Func> resultSelector) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); if (resultSelector == null) throw new ArgumentNullException(nameof(resultSelector)); return Aggregate(source, seed, accumulator, resultSelector, CancellationToken.None); } public static Task Aggregate(this IAsyncEnumerable source, TAccumulate seed, Func> accumulator, Func> resultSelector, CancellationToken cancellationToken) { if (source == null) throw new ArgumentNullException(nameof(source)); if (accumulator == null) throw new ArgumentNullException(nameof(accumulator)); if (resultSelector == null) throw new ArgumentNullException(nameof(resultSelector)); return Aggregate_(source, seed, accumulator, resultSelector, cancellationToken); } private static async Task Aggregate_(IAsyncEnumerable source, TAccumulate seed, Func accumulator, Func resultSelector, CancellationToken cancellationToken) { var acc = seed; var e = source.GetAsyncEnumerator(); try { while (await e.MoveNextAsync(cancellationToken).ConfigureAwait(false)) { acc = accumulator(acc, e.Current); } } finally { await e.DisposeAsync().ConfigureAwait(false); } return resultSelector(acc); } private static async Task Aggregate_(IAsyncEnumerable source, Func accumulator, CancellationToken cancellationToken) { var first = true; var acc = default(TSource); var e = source.GetAsyncEnumerator(); try { while (await e.MoveNextAsync(cancellationToken).ConfigureAwait(false)) { acc = first ? e.Current : accumulator(acc, e.Current); first = false; } } finally { await e.DisposeAsync().ConfigureAwait(false); } if (first) throw new InvalidOperationException(Strings.NO_ELEMENTS); return acc; } private static async Task Aggregate_(IAsyncEnumerable source, TAccumulate seed, Func> accumulator, Func> resultSelector, CancellationToken cancellationToken) { var acc = seed; var e = source.GetAsyncEnumerator(); try { while (await e.MoveNextAsync(cancellationToken).ConfigureAwait(false)) { acc = await accumulator(acc, e.Current).ConfigureAwait(false); } } finally { await e.DisposeAsync().ConfigureAwait(false); } return await resultSelector(acc).ConfigureAwait(false); } private static async Task Aggregate_(IAsyncEnumerable source, Func> accumulator, CancellationToken cancellationToken) { var first = true; var acc = default(TSource); var e = source.GetAsyncEnumerator(); try { while (await e.MoveNextAsync(cancellationToken).ConfigureAwait(false)) { acc = first ? e.Current : await accumulator(acc, e.Current).ConfigureAwait(false); first = false; } } finally { await e.DisposeAsync().ConfigureAwait(false); } if (first) throw new InvalidOperationException(Strings.NO_ELEMENTS); return acc; } } }