// 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 AsyncEnumerableEx { public static IAsyncEnumerable Distinct(this IAsyncEnumerable source, Func keySelector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctCore(source, keySelector, comparer: null); } public static IAsyncEnumerable Distinct(this IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctCore(source, keySelector, comparer); } public static IAsyncEnumerable Distinct(this IAsyncEnumerable source, Func> keySelector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctCore(source, keySelector, comparer: null); } public static IAsyncEnumerable Distinct(this IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); if (comparer == null) throw Error.ArgumentNull(nameof(comparer)); return DistinctCore(source, keySelector, comparer); } private static IAsyncEnumerable DistinctCore(IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { return new DistinctAsyncIterator(source, keySelector, comparer); } private static IAsyncEnumerable DistinctCore(IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { return new DistinctAsyncIteratorWithTask(source, keySelector, comparer); } private sealed class DistinctAsyncIterator : AsyncIterator, IAsyncIListProvider { private readonly IEqualityComparer _comparer; private readonly Func _keySelector; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private Set _set; public DistinctAsyncIterator(IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { Debug.Assert(source != null); Debug.Assert(keySelector != null); _source = source; _keySelector = keySelector; _comparer = comparer; } public async Task ToArrayAsync(CancellationToken cancellationToken) { var s = await FillSetAsync(cancellationToken).ConfigureAwait(false); return s.ToArray(); } public async Task> ToListAsync(CancellationToken cancellationToken) { var s = await FillSetAsync(cancellationToken).ConfigureAwait(false); return s; } public async Task GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) { if (onlyIfCheap) { return -1; } var count = 0; var s = new Set(_comparer); var enu = _source.GetAsyncEnumerator(cancellationToken); try { while (await enu.MoveNextAsync().ConfigureAwait(false)) { var item = enu.Current; if (s.Add(_keySelector(item))) { count++; } } } finally { await enu.DisposeAsync().ConfigureAwait(false); } return count; } public override AsyncIterator Clone() { return new DistinctAsyncIterator(_source, _keySelector, _comparer); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; _set = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); if (!await _enumerator.MoveNextAsync().ConfigureAwait(false)) { await DisposeAsync().ConfigureAwait(false); return false; } var element = _enumerator.Current; _set = new Set(_comparer); _set.Add(_keySelector(element)); _current = element; _state = AsyncIteratorState.Iterating; return true; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { element = _enumerator.Current; if (_set.Add(_keySelector(element))) { _current = element; return true; } } break; } await DisposeAsync().ConfigureAwait(false); return false; } private async Task> FillSetAsync(CancellationToken cancellationToken) { var s = new Set(_comparer); var r = new List(); var enu = _source.GetAsyncEnumerator(cancellationToken); try { while (await enu.MoveNextAsync().ConfigureAwait(false)) { var item = enu.Current; if (s.Add(_keySelector(item))) { r.Add(item); } } } finally { await enu.DisposeAsync().ConfigureAwait(false); } return r; } } private sealed class DistinctAsyncIteratorWithTask : AsyncIterator, IAsyncIListProvider { private readonly IEqualityComparer _comparer; private readonly Func> _keySelector; private readonly IAsyncEnumerable _source; private IAsyncEnumerator _enumerator; private Set _set; public DistinctAsyncIteratorWithTask(IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { Debug.Assert(source != null); Debug.Assert(keySelector != null); _source = source; _keySelector = keySelector; _comparer = comparer; } public async Task ToArrayAsync(CancellationToken cancellationToken) { var s = await FillSetAsync(cancellationToken).ConfigureAwait(false); return s.ToArray(); } public async Task> ToListAsync(CancellationToken cancellationToken) { var s = await FillSetAsync(cancellationToken).ConfigureAwait(false); return s; } public async Task GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) { if (onlyIfCheap) { return -1; } var count = 0; var s = new Set(_comparer); var enu = _source.GetAsyncEnumerator(cancellationToken); try { while (await enu.MoveNextAsync().ConfigureAwait(false)) { var item = enu.Current; if (s.Add(await _keySelector(item).ConfigureAwait(false))) { count++; } } } finally { await enu.DisposeAsync().ConfigureAwait(false); } return count; } public override AsyncIterator Clone() { return new DistinctAsyncIteratorWithTask(_source, _keySelector, _comparer); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; _set = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); if (!await _enumerator.MoveNextAsync().ConfigureAwait(false)) { await DisposeAsync().ConfigureAwait(false); return false; } var element = _enumerator.Current; _set = new Set(_comparer); _set.Add(await _keySelector(element).ConfigureAwait(false)); _current = element; _state = AsyncIteratorState.Iterating; return true; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { element = _enumerator.Current; if (_set.Add(await _keySelector(element).ConfigureAwait(false))) { _current = element; return true; } } break; } await DisposeAsync().ConfigureAwait(false); return false; } private async Task> FillSetAsync(CancellationToken cancellationToken) { var s = new Set(_comparer); var r = new List(); var enu = _source.GetAsyncEnumerator(cancellationToken); try { while (await enu.MoveNextAsync().ConfigureAwait(false)) { var item = enu.Current; if (s.Add(await _keySelector(item).ConfigureAwait(false))) { r.Add(item); } } } finally { await enu.DisposeAsync().ConfigureAwait(false); } return r; } } } }