// 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 TakeWhile(this IAsyncEnumerable source, Func predicate) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); #if CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR // https://github.com/dotnet/roslyn/pull/31114 return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) { if (!predicate(element)) { break; } yield return element; } } #else return new TakeWhileAsyncIterator(source, predicate); #endif } public static IAsyncEnumerable TakeWhile(this IAsyncEnumerable source, Func predicate) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); #if CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR // https://github.com/dotnet/roslyn/pull/31114 return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var index = -1; await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) { checked { index++; } if (!predicate(element, index)) { break; } yield return element; } } #else return new TakeWhileWithIndexAsyncIterator(source, predicate); #endif } public static IAsyncEnumerable TakeWhile(this IAsyncEnumerable source, Func> predicate) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); #if CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR // https://github.com/dotnet/roslyn/pull/31114 return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) { if (!await predicate(element).ConfigureAwait(false)) { break; } yield return element; } } #else return new TakeWhileAsyncIteratorWithTask(source, predicate); #endif } #if !NO_DEEP_CANCELLATION public static IAsyncEnumerable TakeWhile(this IAsyncEnumerable source, Func> predicate) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); #if CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR // https://github.com/dotnet/roslyn/pull/31114 return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) { if (!await predicate(element, cancellationToken).ConfigureAwait(false)) { break; } yield return element; } } #else return new TakeWhileAsyncIteratorWithTaskAndCancellation(source, predicate); #endif } #endif public static IAsyncEnumerable TakeWhile(this IAsyncEnumerable source, Func> predicate) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); #if CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR // https://github.com/dotnet/roslyn/pull/31114 return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var index = -1; await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) { checked { index++; } if (!await predicate(element, index).ConfigureAwait(false)) { break; } yield return element; } } #else return new TakeWhileWithIndexAsyncIteratorWithTask(source, predicate); #endif } #if !NO_DEEP_CANCELLATION public static IAsyncEnumerable TakeWhile(this IAsyncEnumerable source, Func> predicate) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (predicate == null) throw Error.ArgumentNull(nameof(predicate)); #if CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR // https://github.com/dotnet/roslyn/pull/31114 return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var index = -1; await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) { checked { index++; } if (!await predicate(element, index, cancellationToken).ConfigureAwait(false)) { break; } yield return element; } } #else return new TakeWhileWithIndexAsyncIteratorWithTaskAndCancellation(source, predicate); #endif } #endif #if !(CSHARP8 && USE_ASYNC_ITERATOR && ASYNC_ITERATOR_CAN_RETURN_AETOR) private sealed class TakeWhileAsyncIterator : AsyncIterator { private readonly Func _predicate; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; public TakeWhileAsyncIterator(IAsyncEnumerable source, Func predicate) { Debug.Assert(predicate != null); Debug.Assert(source != null); _source = source; _predicate = predicate; } public override AsyncIteratorBase Clone() { return new TakeWhileAsyncIterator(_source, _predicate); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; if (!_predicate(item)) { break; } _current = item; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } } private sealed class TakeWhileWithIndexAsyncIterator : AsyncIterator { private readonly Func _predicate; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private int _index; public TakeWhileWithIndexAsyncIterator(IAsyncEnumerable source, Func predicate) { Debug.Assert(predicate != null); Debug.Assert(source != null); _source = source; _predicate = predicate; } public override AsyncIteratorBase Clone() { return new TakeWhileWithIndexAsyncIterator(_source, _predicate); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _index = -1; _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; checked { _index++; } if (!_predicate(item, _index)) { break; } _current = item; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } } private sealed class TakeWhileAsyncIteratorWithTask : AsyncIterator { private readonly Func> _predicate; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; public TakeWhileAsyncIteratorWithTask(IAsyncEnumerable source, Func> predicate) { Debug.Assert(predicate != null); Debug.Assert(source != null); _source = source; _predicate = predicate; } public override AsyncIteratorBase Clone() { return new TakeWhileAsyncIteratorWithTask(_source, _predicate); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; if (!await _predicate(item).ConfigureAwait(false)) { break; } _current = item; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } } #if !NO_DEEP_CANCELLATION private sealed class TakeWhileAsyncIteratorWithTaskAndCancellation : AsyncIterator { private readonly Func> _predicate; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; public TakeWhileAsyncIteratorWithTaskAndCancellation(IAsyncEnumerable source, Func> predicate) { Debug.Assert(predicate != null); Debug.Assert(source != null); _source = source; _predicate = predicate; } public override AsyncIteratorBase Clone() { return new TakeWhileAsyncIteratorWithTaskAndCancellation(_source, _predicate); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; if (!await _predicate(item, _cancellationToken).ConfigureAwait(false)) { break; } _current = item; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } } #endif private sealed class TakeWhileWithIndexAsyncIteratorWithTask : AsyncIterator { private readonly Func> _predicate; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private int _index; public TakeWhileWithIndexAsyncIteratorWithTask(IAsyncEnumerable source, Func> predicate) { Debug.Assert(predicate != null); Debug.Assert(source != null); _source = source; _predicate = predicate; } public override AsyncIteratorBase Clone() { return new TakeWhileWithIndexAsyncIteratorWithTask(_source, _predicate); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _index = -1; _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; checked { _index++; } if (!await _predicate(item, _index).ConfigureAwait(false)) { break; } _current = item; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } } #if !NO_DEEP_CANCELLATION private sealed class TakeWhileWithIndexAsyncIteratorWithTaskAndCancellation : AsyncIterator { private readonly Func> _predicate; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private int _index; public TakeWhileWithIndexAsyncIteratorWithTaskAndCancellation(IAsyncEnumerable source, Func> predicate) { Debug.Assert(predicate != null); Debug.Assert(source != null); _source = source; _predicate = predicate; } public override AsyncIteratorBase Clone() { return new TakeWhileWithIndexAsyncIteratorWithTaskAndCancellation(_source, _predicate); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _index = -1; _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; checked { _index++; } if (!await _predicate(item, _index, _cancellationToken).ConfigureAwait(false)) { break; } _current = item; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } } #endif #endif } }