// 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 { internal abstract class AsyncIterator : IAsyncEnumerable, IAsyncEnumerator { private readonly int _threadId; private bool _currentIsInvalid = true; internal TSource current; internal AsyncIteratorState state = AsyncIteratorState.New; internal CancellationToken cancellationToken; protected AsyncIterator() { _threadId = Environment.CurrentManagedThreadId; } public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken) { var enumerator = state == AsyncIteratorState.New && _threadId == Environment.CurrentManagedThreadId ? this : Clone(); enumerator.state = AsyncIteratorState.Allocated; enumerator.cancellationToken = cancellationToken; try { enumerator.OnGetEnumerator(cancellationToken); } catch { enumerator.DisposeAsync(); // REVIEW: fire-and-forget? throw; } return enumerator; } public virtual ValueTask DisposeAsync() { current = default; state = AsyncIteratorState.Disposed; return TaskExt.CompletedTask; } public TSource Current { get { if (_currentIsInvalid) throw new InvalidOperationException("Enumerator is in an invalid state"); return current; } } public async ValueTask MoveNextAsync() { // Note: MoveNext *must* be implemented as an async method to ensure // that any exceptions thrown from the MoveNextCore call are handled // by the try/catch, whether they're sync or async if (state == AsyncIteratorState.Disposed) { return false; } try { var result = await MoveNextCore(cancellationToken).ConfigureAwait(false); _currentIsInvalid = !result; // if move next is false, invalid otherwise valid return result; } catch { _currentIsInvalid = true; await DisposeAsync().ConfigureAwait(false); throw; } } public abstract AsyncIterator Clone(); public virtual IAsyncEnumerable Select(Func selector) { return new AsyncEnumerable.SelectEnumerableAsyncIterator(this, selector); } public virtual IAsyncEnumerable Select(Func> selector) { return new AsyncEnumerable.SelectEnumerableAsyncIteratorWithTask(this, selector); } public virtual IAsyncEnumerable Where(Func predicate) { return new AsyncEnumerable.WhereEnumerableAsyncIterator(this, predicate); } public virtual IAsyncEnumerable Where(Func> predicate) { return new AsyncEnumerable.WhereEnumerableAsyncIteratorWithTask(this, predicate); } protected abstract ValueTask MoveNextCore(CancellationToken cancellationToken); protected virtual void OnGetEnumerator(CancellationToken cancellationToken) { } } internal enum AsyncIteratorState { New = 0, Allocated = 1, Iterating = 2, Disposed = -1, } }