// 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 Join(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func resultSelector) => Join(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); public static IAsyncEnumerable Join(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func resultSelector, IEqualityComparer? comparer) { if (outer == null) throw Error.ArgumentNull(nameof(outer)); if (inner == null) throw Error.ArgumentNull(nameof(inner)); if (outerKeySelector == null) throw Error.ArgumentNull(nameof(outerKeySelector)); if (innerKeySelector == null) throw Error.ArgumentNull(nameof(innerKeySelector)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await using (var e = outer.GetConfiguredAsyncEnumerator(cancellationToken, false)) { if (await e.MoveNextAsync()) { var lookup = await Internal.Lookup.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); if (lookup.Count != 0) { do { var item = e.Current; var outerKey = outerKeySelector(item); var g = lookup.GetGrouping(outerKey); if (g != null) { var count = g._count; var elements = g._elements; for (var i = 0; i != count; ++i) { yield return resultSelector(item, elements[i]); } } } while (await e.MoveNextAsync()); } } } } } internal static IAsyncEnumerable JoinAwaitCore(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func> resultSelector) => JoinAwaitCore(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); internal static IAsyncEnumerable JoinAwaitCore(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func> resultSelector, IEqualityComparer? comparer) { if (outer == null) throw Error.ArgumentNull(nameof(outer)); if (inner == null) throw Error.ArgumentNull(nameof(inner)); if (outerKeySelector == null) throw Error.ArgumentNull(nameof(outerKeySelector)); if (innerKeySelector == null) throw Error.ArgumentNull(nameof(innerKeySelector)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await using (var e = outer.GetConfiguredAsyncEnumerator(cancellationToken, false)) { if (await e.MoveNextAsync()) { var lookup = await Internal.LookupWithTask.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); if (lookup.Count != 0) { do { var item = e.Current; var outerKey = await outerKeySelector(item).ConfigureAwait(false); var g = lookup.GetGrouping(outerKey); if (g != null) { var count = g._count; var elements = g._elements; for (var i = 0; i != count; ++i) { yield return await resultSelector(item, elements[i]).ConfigureAwait(false); } } } while (await e.MoveNextAsync()); } } } } } #if !NO_DEEP_CANCELLATION internal static IAsyncEnumerable JoinAwaitWithCancellationCore(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func> resultSelector) => JoinAwaitWithCancellationCore(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); internal static IAsyncEnumerable JoinAwaitWithCancellationCore(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func> resultSelector, IEqualityComparer? comparer) { if (outer == null) throw Error.ArgumentNull(nameof(outer)); if (inner == null) throw Error.ArgumentNull(nameof(inner)); if (outerKeySelector == null) throw Error.ArgumentNull(nameof(outerKeySelector)); if (innerKeySelector == null) throw Error.ArgumentNull(nameof(innerKeySelector)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await using (var e = outer.GetConfiguredAsyncEnumerator(cancellationToken, false)) { if (await e.MoveNextAsync()) { var lookup = await Internal.LookupWithTask.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); if (lookup.Count != 0) { do { var item = e.Current; var outerKey = await outerKeySelector(item, cancellationToken).ConfigureAwait(false); var g = lookup.GetGrouping(outerKey); if (g != null) { var count = g._count; var elements = g._elements; for (var i = 0; i != count; ++i) { yield return await resultSelector(item, elements[i], cancellationToken).ConfigureAwait(false); } } } while (await e.MoveNextAsync()); } } } } } #endif } }