// 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 SkipLast(this IAsyncEnumerable source, int count) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (count <= 0) { // Return source if not actually skipping, but only if it's a type from here, to avoid // issues if collections are used as keys or otherwise must not be aliased. if (source is AsyncIteratorBase) { return source; } count = 0; } #if USE_ASYNC_ITERATOR return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var queue = new Queue(); await using (var e = source.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { while (await e.MoveNextAsync()) { if (queue.Count == count) { do { yield return queue.Dequeue(); queue.Enqueue(e.Current); } while (await e.MoveNextAsync()); break; } else { queue.Enqueue(e.Current); } } } } #else return new SkipLastAsyncIterator(source, count); #endif } #if !USE_ASYNC_ITERATOR private sealed class SkipLastAsyncIterator : AsyncIterator { private readonly int _count; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private Queue _queue; public SkipLastAsyncIterator(IAsyncEnumerable source, int count) { Debug.Assert(source != null); _source = source; _count = count; } public override AsyncIteratorBase Clone() { return new SkipLastAsyncIterator(_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(); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; _queue.Enqueue(item); if (_queue.Count > _count) { _current = _queue.Dequeue(); return true; } } break; } await DisposeAsync().ConfigureAwait(false); return false; } } #endif } }