// 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 Intersect(this IAsyncEnumerable first, IAsyncEnumerable second) { if (first == null) throw Error.ArgumentNull(nameof(first)); if (second == null) throw Error.ArgumentNull(nameof(second)); return first.Intersect(second, EqualityComparer.Default); } public static IAsyncEnumerable Intersect(this IAsyncEnumerable first, IAsyncEnumerable second, IEqualityComparer comparer) { if (first == null) throw Error.ArgumentNull(nameof(first)); if (second == null) throw Error.ArgumentNull(nameof(second)); if (comparer == null) throw Error.ArgumentNull(nameof(comparer)); return new IntersectAsyncIterator(first, second, comparer); } private sealed class IntersectAsyncIterator : AsyncIterator { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _first; private readonly IAsyncEnumerable _second; private Task _fillSetTask; private IAsyncEnumerator _firstEnumerator; private Set _set; private bool _setFilled; public IntersectAsyncIterator(IAsyncEnumerable first, IAsyncEnumerable second, IEqualityComparer comparer) { Debug.Assert(first != null); Debug.Assert(second != null); Debug.Assert(comparer != null); _first = first; _second = second; _comparer = comparer; } public override AsyncIterator Clone() { return new IntersectAsyncIterator(_first, _second, _comparer); } public override async ValueTask DisposeAsync() { if (_firstEnumerator != null) { await _firstEnumerator.DisposeAsync().ConfigureAwait(false); _firstEnumerator = null; } _set = null; await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: _firstEnumerator = _first.GetAsyncEnumerator(cancellationToken); _set = new Set(_comparer); _setFilled = false; _fillSetTask = FillSet(); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: bool moveNext; do { if (!_setFilled) { // This is here so we don't need to call Task.WhenAll each time after the set is filled var moveNextTask = _firstEnumerator.MoveNextAsync(); await Task.WhenAll(moveNextTask.AsTask(), _fillSetTask).ConfigureAwait(false); _setFilled = true; moveNext = moveNextTask.Result; } else { moveNext = await _firstEnumerator.MoveNextAsync().ConfigureAwait(false); } if (moveNext) { var item = _firstEnumerator.Current; if (_set.Remove(item)) { current = item; return true; } } } while (moveNext); await DisposeAsync().ConfigureAwait(false); break; } return false; } private async Task FillSet() { var array = await _second.ToArray().ConfigureAwait(false); for (var i = 0; i < array.Length; i++) { _set.Add(array[i]); } } } } }