// 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 DistinctUntilChanged(this IAsyncEnumerable source) { if (source == null) throw Error.ArgumentNull(nameof(source)); return DistinctUntilChangedCore(source, comparer: null); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, IEqualityComparer comparer) { if (source == null) throw Error.ArgumentNull(nameof(source)); return DistinctUntilChangedCore(source, comparer); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func keySelector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctUntilChangedCore(source, keySelector, comparer: null); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctUntilChangedCore(source, keySelector, comparer); } public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func> keySelector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctUntilChangedCore(source, keySelector, comparer: null); } #if !NO_DEEP_CANCELLATION public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func> keySelector) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctUntilChangedCore(source, keySelector, comparer: null); } #endif public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctUntilChangedCore(source, keySelector, comparer); } #if !NO_DEEP_CANCELLATION public static IAsyncEnumerable DistinctUntilChanged(this IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { if (source == null) throw Error.ArgumentNull(nameof(source)); if (keySelector == null) throw Error.ArgumentNull(nameof(keySelector)); return DistinctUntilChangedCore(source, keySelector, comparer); } #endif private static IAsyncEnumerable DistinctUntilChangedCore(IAsyncEnumerable source, IEqualityComparer comparer) { #if USE_ASYNC_ITERATOR if (comparer == null) { comparer = EqualityComparer.Default; } return AsyncEnumerable.Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false); try // REVIEW: Can use `await using` if we get pattern bind (HAS_AWAIT_USING_PATTERN_BIND) { if (!await e.MoveNextAsync()) { yield break; } var latest = e.Current; yield return latest; while (await e.MoveNextAsync()) { var item = e.Current; if (!comparer.Equals(latest, item)) { latest = item; yield return latest; } } } finally { await e.DisposeAsync(); } } #else return new DistinctUntilChangedAsyncIterator(source, comparer); #endif } private static IAsyncEnumerable DistinctUntilChangedCore(IAsyncEnumerable source, Func keySelector, IEqualityComparer comparer) { #if USE_ASYNC_ITERATOR if (comparer == null) { comparer = EqualityComparer.Default; } return AsyncEnumerable.Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false); try // REVIEW: Can use `await using` if we get pattern bind (HAS_AWAIT_USING_PATTERN_BIND) { if (!await e.MoveNextAsync()) { yield break; } var item = e.Current; var latestKey = keySelector(item); yield return item; while (await e.MoveNextAsync()) { item = e.Current; var currentKey = keySelector(item); if (!comparer.Equals(latestKey, currentKey)) { latestKey = currentKey; yield return item; } } } finally { await e.DisposeAsync(); } } #else return new DistinctUntilChangedAsyncIterator(source, keySelector, comparer); #endif } private static IAsyncEnumerable DistinctUntilChangedCore(IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { #if USE_ASYNC_ITERATOR if (comparer == null) { comparer = EqualityComparer.Default; } return AsyncEnumerable.Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false); try // REVIEW: Can use `await using` if we get pattern bind (HAS_AWAIT_USING_PATTERN_BIND) { if (!await e.MoveNextAsync()) { yield break; } var item = e.Current; var latestKey = await keySelector(item).ConfigureAwait(false); yield return item; while (await e.MoveNextAsync()) { item = e.Current; var currentKey = await keySelector(item).ConfigureAwait(false); if (!comparer.Equals(latestKey, currentKey)) { latestKey = currentKey; yield return item; } } } finally { await e.DisposeAsync(); } } #else return new DistinctUntilChangedAsyncIteratorWithTask(source, keySelector, comparer); #endif } #if !NO_DEEP_CANCELLATION private static IAsyncEnumerable DistinctUntilChangedCore(IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { #if USE_ASYNC_ITERATOR if (comparer == null) { comparer = EqualityComparer.Default; } return AsyncEnumerable.Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { var e = source.GetConfiguredAsyncEnumerator(cancellationToken, false); try // REVIEW: Can use `await using` if we get pattern bind (HAS_AWAIT_USING_PATTERN_BIND) { if (!await e.MoveNextAsync()) { yield break; } var item = e.Current; var latestKey = await keySelector(item, cancellationToken).ConfigureAwait(false); yield return item; while (await e.MoveNextAsync()) { item = e.Current; var currentKey = await keySelector(item, cancellationToken).ConfigureAwait(false); if (!comparer.Equals(latestKey, currentKey)) { latestKey = currentKey; yield return item; } } } finally { await e.DisposeAsync(); } } #else return new DistinctUntilChangedAsyncIteratorWithTaskAndCancellation(source, keySelector, comparer); #endif } #endif #if !USE_ASYNC_ITERATOR 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(source != null); _source = source; _comparer = comparer ?? EqualityComparer.Default; } public override AsyncIteratorBase Clone() { return new DistinctUntilChangedAsyncIterator(_source, _comparer); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; _currentValue = default; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().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; } await DisposeAsync().ConfigureAwait(false); 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) { _source = source; _keySelector = keySelector; _comparer = comparer ?? EqualityComparer.Default; } public override AsyncIteratorBase Clone() { return new DistinctUntilChangedAsyncIterator(_source, _keySelector, _comparer); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; _currentKeyValue = default; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().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 } await DisposeAsync().ConfigureAwait(false); return false; } } private sealed class DistinctUntilChangedAsyncIteratorWithTask : AsyncIterator { private readonly IEqualityComparer _comparer; private readonly Func> _keySelector; private readonly IAsyncEnumerable _source; private TKey _currentKeyValue; private IAsyncEnumerator _enumerator; private bool _hasCurrentKey; public DistinctUntilChangedAsyncIteratorWithTask(IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { _source = source; _keySelector = keySelector; _comparer = comparer ?? EqualityComparer.Default; } public override AsyncIteratorBase Clone() { return new DistinctUntilChangedAsyncIteratorWithTask(_source, _keySelector, _comparer); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; _currentKeyValue = default; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; var key = await _keySelector(item).ConfigureAwait(false); var comparerEquals = false; if (_hasCurrentKey) { comparerEquals = _comparer.Equals(_currentKeyValue, key); } if (!_hasCurrentKey || !comparerEquals) { _hasCurrentKey = true; _currentKeyValue = key; _current = item; return true; } } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } #if !NO_DEEP_CANCELLATION private sealed class DistinctUntilChangedAsyncIteratorWithTaskAndCancellation : AsyncIterator { private readonly IEqualityComparer _comparer; private readonly Func> _keySelector; private readonly IAsyncEnumerable _source; private TKey _currentKeyValue; private IAsyncEnumerator _enumerator; private bool _hasCurrentKey; public DistinctUntilChangedAsyncIteratorWithTaskAndCancellation(IAsyncEnumerable source, Func> keySelector, IEqualityComparer comparer) { _source = source; _keySelector = keySelector; _comparer = comparer ?? EqualityComparer.Default; } public override AsyncIteratorBase Clone() { return new DistinctUntilChangedAsyncIteratorWithTaskAndCancellation(_source, _keySelector, _comparer); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; _currentKeyValue = default; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { var item = _enumerator.Current; var key = await _keySelector(item, _cancellationToken).ConfigureAwait(false); var comparerEquals = false; if (_hasCurrentKey) { comparerEquals = _comparer.Equals(_currentKeyValue, key); } if (!_hasCurrentKey || !comparerEquals) { _hasCurrentKey = true; _currentKeyValue = key; _current = item; return true; } } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } #endif #endif } }