// 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 FirstOrDefaultAsync(this IAsyncEnumerable source, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); return Core(source, cancellationToken); static async Task Core(IAsyncEnumerable _source, CancellationToken _cancellationToken) { var first = await TryGetFirst(_source, _cancellationToken).ConfigureAwait(false); return first.HasValue ? first.Value : default; } } public static Task FirstOrDefaultAsync(this IAsyncEnumerable source, Func predicate, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); return Core(source, predicate, cancellationToken); static async Task Core(IAsyncEnumerable _source, Func _predicate, CancellationToken _cancellationToken) { var first = await TryGetFirst(_source, _predicate, _cancellationToken).ConfigureAwait(false); return first.HasValue ? first.Value : default; } } public static Task FirstOrDefaultAsync(this IAsyncEnumerable source, Func> predicate, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); return Core(source, predicate, cancellationToken); static async Task Core(IAsyncEnumerable _source, Func> _predicate, CancellationToken _cancellationToken) { var first = await TryGetFirst(_source, _predicate, _cancellationToken).ConfigureAwait(false); return first.HasValue ? first.Value : default; } } #if !NO_DEEP_CANCELLATION public static Task FirstOrDefaultAsync(this IAsyncEnumerable source, Func> predicate, CancellationToken cancellationToken = default) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); return Core(source, predicate, cancellationToken); static async Task Core(IAsyncEnumerable _source, Func> _predicate, CancellationToken _cancellationToken) { var first = await TryGetFirst(_source, _predicate, _cancellationToken).ConfigureAwait(false); return first.HasValue ? first.Value : default; } } #endif private static ValueTask> TryGetFirst(IAsyncEnumerable source, CancellationToken cancellationToken) { if (source is IList list) { if (list.Count > 0) { return new ValueTask>(new Maybe(list[0])); } } else if (source is IAsyncPartition p) { return p.TryGetFirstAsync(cancellationToken); } else { return Core(source, cancellationToken); static async ValueTask> Core(IAsyncEnumerable _source, CancellationToken _cancellationToken) { #if USE_AWAIT_USING await using (var e = _source.GetAsyncEnumerator(_cancellationToken).ConfigureAwait(false)) { if (await e.MoveNextAsync()) { return new Maybe(e.Current); } } #else var e = _source.GetAsyncEnumerator(_cancellationToken); try { if (await e.MoveNextAsync().ConfigureAwait(false)) { return new Maybe(e.Current); } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return new Maybe(); } } return new ValueTask>(new Maybe()); } private static async Task> TryGetFirst(IAsyncEnumerable source, Func predicate, CancellationToken cancellationToken) { #if USE_AWAIT_USING await using (var e = source.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { while (await e.MoveNextAsync()) { var value = e.Current; if (predicate(value)) { return new Maybe(value); } } } #else var e = source.GetAsyncEnumerator(cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { var value = e.Current; if (predicate(value)) { return new Maybe(value); } } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return new Maybe(); } private static async Task> TryGetFirst(IAsyncEnumerable source, Func> predicate, CancellationToken cancellationToken) { #if USE_AWAIT_USING await using (var e = source.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { while (await e.MoveNextAsync()) { var value = e.Current; if (await predicate(value).ConfigureAwait(false)) { return new Maybe(value); } } } #else var e = source.GetAsyncEnumerator(cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { var value = e.Current; if (await predicate(value).ConfigureAwait(false)) { return new Maybe(value); } } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return new Maybe(); } #if !NO_DEEP_CANCELLATION private static async Task> TryGetFirst(IAsyncEnumerable source, Func> predicate, CancellationToken cancellationToken) { #if USE_AWAIT_USING await using (var e = source.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { while (await e.MoveNextAsync()) { var value = e.Current; if (await predicate(value, cancellationToken).ConfigureAwait(false)) { return new Maybe(value); } } } #else var e = source.GetAsyncEnumerator(cancellationToken); try { while (await e.MoveNextAsync().ConfigureAwait(false)) { var value = e.Current; if (await predicate(value, cancellationToken).ConfigureAwait(false)) { return new Maybe(value); } } } finally { await e.DisposeAsync().ConfigureAwait(false); } #endif return new Maybe(); } #endif } }