// 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.Diagnostics; using System.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { public static IAsyncEnumerable Where(this IAsyncEnumerable source, Func predicate) { if (source == null) throw new ArgumentNullException(nameof(source)); if (predicate == null) throw new ArgumentNullException(nameof(predicate)); var iterator = source as AsyncIterator; if (iterator != null) { return iterator.Where(predicate); } // TODO: Can we add array/list optimizations here, does it make sense? return new WhereEnumerableAsyncIterator(source, predicate); } public static IAsyncEnumerable Where(this IAsyncEnumerable source, Func predicate) { if (source == null) throw new ArgumentNullException(nameof(source)); if (predicate == null) throw new ArgumentNullException(nameof(predicate)); return new WhereEnumerableWithIndexAsyncIterator(source, predicate); } private static Func CombinePredicates(Func predicate1, Func predicate2) { return x => predicate1(x) && predicate2(x); } internal sealed class WhereEnumerableAsyncIterator : AsyncIterator { private readonly Func predicate; private readonly IAsyncEnumerable source; private IAsyncEnumerator enumerator; public WhereEnumerableAsyncIterator(IAsyncEnumerable source, Func predicate) { Debug.Assert(source != null); Debug.Assert(predicate != null); this.source = source; this.predicate = predicate; } public override AsyncIterator Clone() { return new WhereEnumerableAsyncIterator(source, predicate); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; } base.Dispose(); } public override IAsyncEnumerable Select(Func selector) { return new WhereSelectEnumerableAsyncIterator(source, predicate, selector); } public override IAsyncEnumerable Where(Func predicate) { return new WhereEnumerableAsyncIterator(source, CombinePredicates(this.predicate, predicate)); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { var item = enumerator.Current; if (predicate(item)) { current = item; return true; } } Dispose(); break; } return false; } } internal sealed class WhereEnumerableWithIndexAsyncIterator : AsyncIterator { private readonly Func predicate; private readonly IAsyncEnumerable source; private IAsyncEnumerator enumerator; private int index; public WhereEnumerableWithIndexAsyncIterator(IAsyncEnumerable source, Func predicate) { Debug.Assert(source != null); Debug.Assert(predicate != null); this.source = source; this.predicate = predicate; } public override AsyncIterator Clone() { return new WhereEnumerableWithIndexAsyncIterator(source, predicate); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; } base.Dispose(); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); index = -1; state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { checked { index++; } var item = enumerator.Current; if (predicate(item, index)) { current = item; return true; } } Dispose(); break; } return false; } } internal sealed class WhereSelectEnumerableAsyncIterator : AsyncIterator { private readonly Func predicate; private readonly Func selector; private readonly IAsyncEnumerable source; private IAsyncEnumerator enumerator; public WhereSelectEnumerableAsyncIterator(IAsyncEnumerable source, Func predicate, Func selector) { Debug.Assert(source != null); Debug.Assert(predicate != null); Debug.Assert(selector != null); this.source = source; this.predicate = predicate; this.selector = selector; } public override AsyncIterator Clone() { return new WhereSelectEnumerableAsyncIterator(source, predicate, selector); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; } base.Dispose(); } public override IAsyncEnumerable Select(Func selector) { return new WhereSelectEnumerableAsyncIterator(source, predicate, CombineSelectors(this.selector, selector)); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { var item = enumerator.Current; if (predicate(item)) { current = selector(item); return true; } } Dispose(); break; } return false; } } } }