// 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 TakeLast(this IAsyncEnumerable source, int count) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (count <= 0) { return Empty(); } #if USE_ASYNC_ITERATOR return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { Queue queue; await using (var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false)) { if (!await e.MoveNextAsync()) { yield break; } queue = new Queue(); queue.Enqueue(e.Current); while (await e.MoveNextAsync()) { if (queue.Count < count) { queue.Enqueue(e.Current); } else { do { queue.Dequeue(); queue.Enqueue(e.Current); } while (await e.MoveNextAsync()); break; } } } Debug.Assert(queue.Count <= count); do { yield return queue.Dequeue(); } while (queue.Count > 0); } #else return new TakeLastAsyncIterator(source, count); #endif } #if !USE_ASYNC_ITERATOR private sealed class TakeLastAsyncIterator : AsyncIterator { private readonly int _count; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private bool _isDone; private Queue _queue; public TakeLastAsyncIterator(IAsyncEnumerable source, int count) { Debug.Assert(source != null); _source = source; _count = count; } public override AsyncIteratorBase Clone() { return new TakeLastAsyncIterator(_source, _count); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } _queue = null; // release the memory await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _queue = new Queue(); _isDone = false; _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (true) { if (!_isDone) { if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { if (_count > 0) { var item = _enumerator.Current; if (_queue.Count >= _count) { _queue.Dequeue(); } _queue.Enqueue(item); } } else { _isDone = true; // Dispose early here as we can await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } continue; // loop until queue is drained } if (_queue.Count > 0) { _current = _queue.Dequeue(); return true; } break; // while } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } #endif } }