// 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.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { public static IAsyncEnumerable TakeLast(this IAsyncEnumerable source, int count) { if (source == null) throw new ArgumentNullException(nameof(source)); if (count <= 0) { return Empty(); } return new TakeLastAsyncIterator(source, count); } 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); this.source = source; this.count = count; } public override AsyncIterator Clone() { return new TakeLastAsyncIterator(source, count); } public override async Task DisposeAsync() { if (enumerator != null) { await enumerator.DisposeAsync().ConfigureAwait(false); enumerator = null; } queue = null; // release the memory await base.DisposeAsync().ConfigureAwait(false); } protected override async Task MoveNextCore() { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetAsyncEnumerator(); 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; } } } }