Join.cs 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177
  1. // Licensed to the .NET Foundation under one or more agreements.
  2. // The .NET Foundation licenses this file to you under the Apache 2.0 License.
  3. // See the LICENSE file in the project root for more information.
  4. using System.Collections.Generic;
  5. using System.Threading;
  6. using System.Threading.Tasks;
  7. namespace System.Linq
  8. {
  9. public static partial class AsyncEnumerable
  10. {
  11. public static IAsyncEnumerable<TResult> Join<TOuter, TInner, TKey, TResult>(this IAsyncEnumerable<TOuter> outer, IAsyncEnumerable<TInner> inner, Func<TOuter, TKey> outerKeySelector, Func<TInner, TKey> innerKeySelector, Func<TOuter, TInner, TResult> resultSelector) =>
  12. Join(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null);
  13. public static IAsyncEnumerable<TResult> Join<TOuter, TInner, TKey, TResult>(this IAsyncEnumerable<TOuter> outer, IAsyncEnumerable<TInner> inner, Func<TOuter, TKey> outerKeySelector, Func<TInner, TKey> innerKeySelector, Func<TOuter, TInner, TResult> resultSelector, IEqualityComparer<TKey>? comparer)
  14. {
  15. if (outer == null)
  16. throw Error.ArgumentNull(nameof(outer));
  17. if (inner == null)
  18. throw Error.ArgumentNull(nameof(inner));
  19. if (outerKeySelector == null)
  20. throw Error.ArgumentNull(nameof(outerKeySelector));
  21. if (innerKeySelector == null)
  22. throw Error.ArgumentNull(nameof(innerKeySelector));
  23. if (resultSelector == null)
  24. throw Error.ArgumentNull(nameof(resultSelector));
  25. return Create(Core);
  26. async IAsyncEnumerator<TResult> Core(CancellationToken cancellationToken)
  27. {
  28. await using (var e = outer.GetConfiguredAsyncEnumerator(cancellationToken, false))
  29. {
  30. if (await e.MoveNextAsync())
  31. {
  32. var lookup = await Internal.Lookup<TKey, TInner>.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false);
  33. if (lookup.Count != 0)
  34. {
  35. do
  36. {
  37. var item = e.Current;
  38. var outerKey = outerKeySelector(item);
  39. var g = lookup.GetGrouping(outerKey);
  40. if (g != null)
  41. {
  42. var count = g._count;
  43. var elements = g._elements;
  44. for (var i = 0; i != count; ++i)
  45. {
  46. yield return resultSelector(item, elements[i]);
  47. }
  48. }
  49. }
  50. while (await e.MoveNextAsync());
  51. }
  52. }
  53. }
  54. }
  55. }
  56. internal static IAsyncEnumerable<TResult> JoinAwaitCore<TOuter, TInner, TKey, TResult>(this IAsyncEnumerable<TOuter> outer, IAsyncEnumerable<TInner> inner, Func<TOuter, ValueTask<TKey>> outerKeySelector, Func<TInner, ValueTask<TKey>> innerKeySelector, Func<TOuter, TInner, ValueTask<TResult>> resultSelector) =>
  57. JoinAwaitCore<TOuter, TInner, TKey, TResult>(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null);
  58. internal static IAsyncEnumerable<TResult> JoinAwaitCore<TOuter, TInner, TKey, TResult>(this IAsyncEnumerable<TOuter> outer, IAsyncEnumerable<TInner> inner, Func<TOuter, ValueTask<TKey>> outerKeySelector, Func<TInner, ValueTask<TKey>> innerKeySelector, Func<TOuter, TInner, ValueTask<TResult>> resultSelector, IEqualityComparer<TKey>? comparer)
  59. {
  60. if (outer == null)
  61. throw Error.ArgumentNull(nameof(outer));
  62. if (inner == null)
  63. throw Error.ArgumentNull(nameof(inner));
  64. if (outerKeySelector == null)
  65. throw Error.ArgumentNull(nameof(outerKeySelector));
  66. if (innerKeySelector == null)
  67. throw Error.ArgumentNull(nameof(innerKeySelector));
  68. if (resultSelector == null)
  69. throw Error.ArgumentNull(nameof(resultSelector));
  70. return Create(Core);
  71. async IAsyncEnumerator<TResult> Core(CancellationToken cancellationToken)
  72. {
  73. await using (var e = outer.GetConfiguredAsyncEnumerator(cancellationToken, false))
  74. {
  75. if (await e.MoveNextAsync())
  76. {
  77. var lookup = await Internal.LookupWithTask<TKey, TInner>.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false);
  78. if (lookup.Count != 0)
  79. {
  80. do
  81. {
  82. var item = e.Current;
  83. var outerKey = await outerKeySelector(item).ConfigureAwait(false);
  84. var g = lookup.GetGrouping(outerKey);
  85. if (g != null)
  86. {
  87. var count = g._count;
  88. var elements = g._elements;
  89. for (var i = 0; i != count; ++i)
  90. {
  91. yield return await resultSelector(item, elements[i]).ConfigureAwait(false);
  92. }
  93. }
  94. }
  95. while (await e.MoveNextAsync());
  96. }
  97. }
  98. }
  99. }
  100. }
  101. #if !NO_DEEP_CANCELLATION
  102. internal static IAsyncEnumerable<TResult> JoinAwaitWithCancellationCore<TOuter, TInner, TKey, TResult>(this IAsyncEnumerable<TOuter> outer, IAsyncEnumerable<TInner> inner, Func<TOuter, CancellationToken, ValueTask<TKey>> outerKeySelector, Func<TInner, CancellationToken, ValueTask<TKey>> innerKeySelector, Func<TOuter, TInner, CancellationToken, ValueTask<TResult>> resultSelector) =>
  103. JoinAwaitWithCancellationCore<TOuter, TInner, TKey, TResult>(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null);
  104. internal static IAsyncEnumerable<TResult> JoinAwaitWithCancellationCore<TOuter, TInner, TKey, TResult>(this IAsyncEnumerable<TOuter> outer, IAsyncEnumerable<TInner> inner, Func<TOuter, CancellationToken, ValueTask<TKey>> outerKeySelector, Func<TInner, CancellationToken, ValueTask<TKey>> innerKeySelector, Func<TOuter, TInner, CancellationToken, ValueTask<TResult>> resultSelector, IEqualityComparer<TKey>? comparer)
  105. {
  106. if (outer == null)
  107. throw Error.ArgumentNull(nameof(outer));
  108. if (inner == null)
  109. throw Error.ArgumentNull(nameof(inner));
  110. if (outerKeySelector == null)
  111. throw Error.ArgumentNull(nameof(outerKeySelector));
  112. if (innerKeySelector == null)
  113. throw Error.ArgumentNull(nameof(innerKeySelector));
  114. if (resultSelector == null)
  115. throw Error.ArgumentNull(nameof(resultSelector));
  116. return Create(Core);
  117. async IAsyncEnumerator<TResult> Core(CancellationToken cancellationToken)
  118. {
  119. await using (var e = outer.GetConfiguredAsyncEnumerator(cancellationToken, false))
  120. {
  121. if (await e.MoveNextAsync())
  122. {
  123. var lookup = await Internal.LookupWithTask<TKey, TInner>.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false);
  124. if (lookup.Count != 0)
  125. {
  126. do
  127. {
  128. var item = e.Current;
  129. var outerKey = await outerKeySelector(item, cancellationToken).ConfigureAwait(false);
  130. var g = lookup.GetGrouping(outerKey);
  131. if (g != null)
  132. {
  133. var count = g._count;
  134. var elements = g._elements;
  135. for (var i = 0; i != count; ++i)
  136. {
  137. yield return await resultSelector(item, elements[i], cancellationToken).ConfigureAwait(false);
  138. }
  139. }
  140. }
  141. while (await e.MoveNextAsync());
  142. }
  143. }
  144. }
  145. }
  146. }
  147. #endif
  148. }
  149. }