// 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.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { /// /// Correlates the elements of two sequences based on matching keys. The default equality comparer is used to compare keys. /// /// The type of the elements of the first async-enumerable sequence. /// The type of the elements of the second async-enumerable sequence. /// The type of the keys returned by the key selector functions. /// The type of the result elements. /// The first async-enumerable sequence to join. /// The async-enumerable sequence to join to the first sequence. /// A function to extract the join key from each element of the first sequence. /// A function to extract the join key from each element of the second sequence. /// A function to create a result element from two matching elements. /// An async-enumerable sequence that has elements of type TResult that are obtained by performing an inner join on two sequences. /// or or or or is null. public static IAsyncEnumerable Join(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func resultSelector) => Join(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); /// /// Correlates the elements of two sequences based on matching keys. A specified equality comparer is used to compare keys. /// /// The type of the elements of the first async-enumerable sequence. /// The type of the elements of the second async-enumerable sequence. /// The type of the keys returned by the key selector functions. /// The type of the result elements. /// The first async-enumerable sequence to join. /// The async-enumerable sequence to join to the first sequence. /// A function to extract the join key from each element of the first sequence. /// A function to extract the join key from each element of the second sequence. /// A function to create a result element from two matching elements. /// An equality comparer to hash and compare keys. /// An async-enumerable sequence that has elements of type TResult that are obtained by performing an inner join on two sequences. /// or or or or is 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 } }