// 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; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { public static IAsyncEnumerable Intersect(this IAsyncEnumerable first, IAsyncEnumerable second, IEqualityComparer comparer) { if (first == null) throw new ArgumentNullException(nameof(first)); if (second == null) throw new ArgumentNullException(nameof(second)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); return CreateEnumerable( () => { var e = first.GetEnumerator(); var cts = new CancellationTokenDisposable(); var d = Disposable.Create(cts, e); var mapTask = default(Task>); var getMapTask = new Func>>( ct => { if (mapTask == null) mapTask = second.ToDictionary(x => x, comparer, ct); return mapTask; }); var f = default(Func>); f = async ct => { if (await e.MoveNext(ct) .Zip(getMapTask(ct), (b, _) => b) .ConfigureAwait(false)) { // Note: Result here is safe because the task // was completed in the Zip() call above if (mapTask.Result.ContainsKey(e.Current)) return true; return await f(ct) .ConfigureAwait(false); } return false; }; return CreateEnumerator( f, () => e.Current, d.Dispose, e ); }); } public static IAsyncEnumerable Intersect(this IAsyncEnumerable first, IAsyncEnumerable second) { if (first == null) throw new ArgumentNullException(nameof(first)); if (second == null) throw new ArgumentNullException(nameof(second)); return first.Intersect(second, EqualityComparer.Default); } } }