// 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 Distinct(this IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { if (source == null) throw new ArgumentNullException(nameof(source)); if (keySelector == null) throw new ArgumentNullException(nameof(keySelector)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); return new DistinctAsyncIterator(source, keySelector, comparer); } public static IAsyncEnumerable Distinct(this IAsyncEnumerable source, Func keySelector) { if (source == null) throw new ArgumentNullException(nameof(source)); if (keySelector == null) throw new ArgumentNullException(nameof(keySelector)); return source.Distinct(keySelector, EqualityComparer.Default); } public static IAsyncEnumerable Distinct(this IAsyncEnumerable source, IEqualityComparer comparer) { if (source == null) throw new ArgumentNullException(nameof(source)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); return new DistinctAsyncIterator(source, comparer); } public static IAsyncEnumerable Distinct(this IAsyncEnumerable source) { if (source == null) throw new ArgumentNullException(nameof(source)); return source.Distinct(EqualityComparer.Default); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source) { if (source == null) throw new ArgumentNullException(nameof(source)); return source.DistinctUntilChanged(EqualityComparer.Default); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, IEqualityComparer comparer) { if (source == null) throw new ArgumentNullException(nameof(source)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); return new DistinctUntilChangedAsyncIterator(source, comparer); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func keySelector) { if (source == null) throw new ArgumentNullException(nameof(source)); if (keySelector == null) throw new ArgumentNullException(nameof(keySelector)); return source.DistinctUntilChanged_(keySelector, EqualityComparer.Default); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { if (source == null) throw new ArgumentNullException(nameof(source)); if (keySelector == null) throw new ArgumentNullException(nameof(keySelector)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); return source.DistinctUntilChanged_(keySelector, comparer); } private static IAsyncEnumerable DistinctUntilChanged_(this IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { return new DistinctUntilChangedAsyncIterator(source, keySelector, comparer); } private sealed class DistinctAsyncIterator : AsyncIterator, IIListProvider { 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); Debug.Assert(comparer != null); this.source = source; this.keySelector = keySelector; this.comparer = comparer; } public async Task ToArrayAsync(CancellationToken cancellationToken) { var s = await FillSet(cancellationToken) .ConfigureAwait(false); return s.ToArray(); } public async Task> ToListAsync(CancellationToken cancellationToken) { var s = await FillSet(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); using (var enu = source.GetEnumerator()) { while (await enu.MoveNext(cancellationToken) .ConfigureAwait(false)) { var item = enu.Current; if (s.Add(keySelector(item))) { count++; } } } return count; } public override AsyncIterator Clone() { return new DistinctAsyncIterator(source, keySelector, comparer); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; set = null; } base.Dispose(); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); if (!await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { Dispose(); 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.MoveNext(cancellationToken) .ConfigureAwait(false)) { element = enumerator.Current; if (set.Add(keySelector(element))) { current = element; return true; } } break; } Dispose(); return false; } private async Task> FillSet(CancellationToken cancellationToken) { var s = new Set(comparer); var r = new List(); using (var enu = source.GetEnumerator()) { while (await enu.MoveNext(cancellationToken) .ConfigureAwait(false)) { var item = enu.Current; if (s.Add(keySelector(item))) { r.Add(item); } } } return r; } } private sealed class DistinctAsyncIterator : AsyncIterator, IIListProvider { private readonly IEqualityComparer comparer; private readonly IAsyncEnumerable source; private IAsyncEnumerator enumerator; private Set set; public DistinctAsyncIterator(IAsyncEnumerable source, IEqualityComparer comparer) { Debug.Assert(source != null); this.source = source; this.comparer = comparer; } public async Task ToArrayAsync(CancellationToken cancellationToken) { var s = await FillSet(cancellationToken) .ConfigureAwait(false); return s.ToArray(); } public async Task> ToListAsync(CancellationToken cancellationToken) { var s = await FillSet(cancellationToken) .ConfigureAwait(false); return s.ToList(); } public async Task GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) { return onlyIfCheap ? -1 : (await FillSet(cancellationToken) .ConfigureAwait(false)).Count; } public override AsyncIterator Clone() { return new DistinctAsyncIterator(source, comparer); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; set = null; } base.Dispose(); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); if (!await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { Dispose(); return false; } var element = enumerator.Current; set = new Set(comparer); set.Add(element); current = element; state = AsyncIteratorState.Iterating; return true; case AsyncIteratorState.Iterating: while (await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { element = enumerator.Current; if (set.Add(element)) { current = element; return true; } } break; } Dispose(); return false; } private async Task> FillSet(CancellationToken cancellationToken) { var s = new Set(comparer); using (var enu = source.GetEnumerator()) { while (await enu.MoveNext(cancellationToken) .ConfigureAwait(false)) { s.Add(enu.Current); } } return s; } } private sealed class DistinctUntilChangedAsyncIterator : AsyncIterator { private readonly IEqualityComparer comparer; private readonly IAsyncEnumerable source; private TSource currentValue; private IAsyncEnumerator enumerator; private bool hasCurrentValue; public DistinctUntilChangedAsyncIterator(IAsyncEnumerable source, IEqualityComparer comparer) { Debug.Assert(comparer != null); Debug.Assert(source != null); this.source = source; this.comparer = comparer; } public override AsyncIterator Clone() { return new DistinctUntilChangedAsyncIterator(source, comparer); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; currentValue = default(TSource); } base.Dispose(); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { var item = enumerator.Current; var comparerEquals = false; if (hasCurrentValue) { comparerEquals = comparer.Equals(currentValue, item); } if (!hasCurrentValue || !comparerEquals) { hasCurrentValue = true; currentValue = item; current = item; return true; } } break; } Dispose(); return false; } } private sealed class DistinctUntilChangedAsyncIterator : AsyncIterator { private readonly IEqualityComparer comparer; private readonly Func keySelector; private readonly IAsyncEnumerable source; private TKey currentKeyValue; private IAsyncEnumerator enumerator; private bool hasCurrentKey; public DistinctUntilChangedAsyncIterator(IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { this.source = source; this.keySelector = keySelector; this.comparer = comparer; } public override AsyncIterator Clone() { return new DistinctUntilChangedAsyncIterator(source, keySelector, comparer); } public override void Dispose() { if (enumerator != null) { enumerator.Dispose(); enumerator = null; currentKeyValue = default(TKey); } base.Dispose(); } protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: enumerator = source.GetEnumerator(); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await enumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { var item = enumerator.Current; var key = keySelector(item); var comparerEquals = false; if (hasCurrentKey) { comparerEquals = comparer.Equals(currentKeyValue, key); } if (!hasCurrentKey || !comparerEquals) { hasCurrentKey = true; currentKeyValue = key; current = item; return true; } } break; // case } Dispose(); return false; } } } }