SelectMany.cs 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817
  1. // Licensed to the .NET Foundation under one or more agreements.
  2. // The .NET Foundation licenses this file to you under the MIT 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. /// <summary>
  12. /// Projects each element of an async-enumerable sequence to an async-enumerable sequence and merges the resulting async-enumerable sequences into one async-enumerable sequence.
  13. /// </summary>
  14. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  15. /// <typeparam name="TResult">The type of the elements in the projected inner sequences and the elements in the merged result sequence.</typeparam>
  16. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  17. /// <param name="selector">A transform function to apply to each element.</param>
  18. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function on each element of the input sequence.</returns>
  19. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="selector"/> is null.</exception>
  20. public static IAsyncEnumerable<TResult> SelectMany<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TResult>> selector)
  21. {
  22. if (source == null)
  23. throw Error.ArgumentNull(nameof(source));
  24. if (selector == null)
  25. throw Error.ArgumentNull(nameof(selector));
  26. return new SelectManyAsyncIterator<TSource, TResult>(source, selector);
  27. }
  28. // REVIEW: Should we keep these overloads that return ValueTask<IAsyncEnumerable<TResult>>? One could argue the selector is async twice.
  29. /// <summary>
  30. /// Projects each element of an async-enumerable sequence into an async-enumerable sequence and merges the resulting async-enumerable sequences into one async-enumerable sequence.
  31. /// </summary>
  32. /// <typeparam name="TSource">The type of elements in the source sequence.</typeparam>
  33. /// <typeparam name="TResult">The type of elements in the projected inner sequences and the merged result sequence.</typeparam>
  34. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  35. /// <param name="selector">An asynchronous selector function to apply to each element of the source sequence.</param>
  36. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function on each element of the source sequence and awaiting the result.</returns>
  37. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="selector"/> is null.</exception>
  38. [GenerateAsyncOverload]
  39. private static IAsyncEnumerable<TResult> SelectManyAwaitCore<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, ValueTask<IAsyncEnumerable<TResult>>> selector)
  40. {
  41. if (source == null)
  42. throw Error.ArgumentNull(nameof(source));
  43. if (selector == null)
  44. throw Error.ArgumentNull(nameof(selector));
  45. return new SelectManyAsyncIteratorWithTask<TSource, TResult>(source, selector);
  46. }
  47. #if !NO_DEEP_CANCELLATION
  48. [GenerateAsyncOverload]
  49. private static IAsyncEnumerable<TResult> SelectManyAwaitWithCancellationCore<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, ValueTask<IAsyncEnumerable<TResult>>> selector)
  50. {
  51. if (source == null)
  52. throw Error.ArgumentNull(nameof(source));
  53. if (selector == null)
  54. throw Error.ArgumentNull(nameof(selector));
  55. return new SelectManyAsyncIteratorWithTaskAndCancellation<TSource, TResult>(source, selector);
  56. }
  57. #endif
  58. /// <summary>
  59. /// Projects each element of an async-enumerable sequence to an async-enumerable sequence by incorporating the element's index and merges the resulting async-enumerable sequences into one async-enumerable sequence.
  60. /// </summary>
  61. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  62. /// <typeparam name="TResult">The type of the elements in the projected inner sequences and the elements in the merged result sequence.</typeparam>
  63. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  64. /// <param name="selector">A transform function to apply to each element; the second parameter of the function represents the index of the source element.</param>
  65. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function on each element of the input sequence.</returns>
  66. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="selector"/> is null.</exception>
  67. public static IAsyncEnumerable<TResult> SelectMany<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TResult>> selector)
  68. {
  69. if (source == null)
  70. throw Error.ArgumentNull(nameof(source));
  71. if (selector == null)
  72. throw Error.ArgumentNull(nameof(selector));
  73. return Core(source, selector);
  74. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TResult>> selector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  75. {
  76. var index = -1;
  77. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  78. {
  79. checked
  80. {
  81. index++;
  82. }
  83. var inner = selector(element, index);
  84. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  85. {
  86. yield return subElement;
  87. }
  88. }
  89. }
  90. }
  91. /// <summary>
  92. /// Projects each element of an async-enumerable sequence into an async-enumerable sequence by incorporating the element's index and merges the resulting async-enumerable sequences into an async-enumerable sequence.
  93. /// </summary>
  94. /// <typeparam name="TSource">The type of elements in the source sequence.</typeparam>
  95. /// <typeparam name="TResult">The type of elements in the projected inner sequences and the merged result sequence.</typeparam>
  96. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  97. /// <param name="selector">An asynchronous selector function to apply to each element; the second parameter represents the index of the element.</param>
  98. /// <returns>An async-enumerable sequence who's elements are the result of invoking the one-to-many transform function on each element of the source sequence and awaiting the result.</returns>
  99. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="selector"/> is null.</exception>
  100. [GenerateAsyncOverload]
  101. private static IAsyncEnumerable<TResult> SelectManyAwaitCore<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, ValueTask<IAsyncEnumerable<TResult>>> selector)
  102. {
  103. if (source == null)
  104. throw Error.ArgumentNull(nameof(source));
  105. if (selector == null)
  106. throw Error.ArgumentNull(nameof(selector));
  107. return Core(source, selector);
  108. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, int, ValueTask<IAsyncEnumerable<TResult>>> selector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  109. {
  110. var index = -1;
  111. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  112. {
  113. checked
  114. {
  115. index++;
  116. }
  117. var inner = await selector(element, index).ConfigureAwait(false);
  118. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  119. {
  120. yield return subElement;
  121. }
  122. }
  123. }
  124. }
  125. #if !NO_DEEP_CANCELLATION
  126. [GenerateAsyncOverload]
  127. private static IAsyncEnumerable<TResult> SelectManyAwaitWithCancellationCore<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, CancellationToken, ValueTask<IAsyncEnumerable<TResult>>> selector)
  128. {
  129. if (source == null)
  130. throw Error.ArgumentNull(nameof(source));
  131. if (selector == null)
  132. throw Error.ArgumentNull(nameof(selector));
  133. return Core(source, selector);
  134. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, int, CancellationToken, ValueTask<IAsyncEnumerable<TResult>>> selector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  135. {
  136. var index = -1;
  137. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  138. {
  139. checked
  140. {
  141. index++;
  142. }
  143. var inner = await selector(element, index, cancellationToken).ConfigureAwait(false);
  144. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  145. {
  146. yield return subElement;
  147. }
  148. }
  149. }
  150. }
  151. #endif
  152. /// <summary>
  153. /// Projects each element of an async-enumerable sequence to an async-enumerable sequence, invokes the result selector for the source element and each of the corresponding inner sequence's elements, and merges the results into one async-enumerable sequence.
  154. /// </summary>
  155. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  156. /// <typeparam name="TCollection">The type of the elements in the projected intermediate sequences.</typeparam>
  157. /// <typeparam name="TResult">The type of the elements in the result sequence, obtained by using the selector to combine source sequence elements with their corresponding intermediate sequence elements.</typeparam>
  158. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  159. /// <param name="collectionSelector">A transform function to apply to each element.</param>
  160. /// <param name="resultSelector">A transform function to apply to each element of the intermediate sequence.</param>
  161. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function collectionSelector on each element of the input sequence and then mapping each of those sequence elements and their corresponding source element to a result element.</returns>
  162. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="collectionSelector"/> or <paramref name="resultSelector"/> is null.</exception>
  163. public static IAsyncEnumerable<TResult> SelectMany<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector)
  164. {
  165. if (source == null)
  166. throw Error.ArgumentNull(nameof(source));
  167. if (collectionSelector == null)
  168. throw Error.ArgumentNull(nameof(collectionSelector));
  169. if (resultSelector == null)
  170. throw Error.ArgumentNull(nameof(resultSelector));
  171. return Core(source, collectionSelector, resultSelector);
  172. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  173. {
  174. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  175. {
  176. var inner = collectionSelector(element);
  177. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  178. {
  179. yield return resultSelector(element, subElement);
  180. }
  181. }
  182. }
  183. }
  184. /// <summary>
  185. /// Projects each element of an async-enumerable sequence to an async-enumerable sequence by awaiting the result of a transform function, invokes the result selector for each of the source elements and each of the corrasponding inner-sequence's elements and awaits the result, and merges the results into one async-enumerable sequence.
  186. /// </summary>
  187. /// <typeparam name="TSource">The type of elements in the source sequence.</typeparam>
  188. /// <typeparam name="TCollection">The type of elements in the projected intermediate sequences.</typeparam>
  189. /// <typeparam name="TResult">The type of elements in the result sequence.</typeparam>
  190. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  191. /// <param name="collectionSelector">An asynchronous transform function to apply to each source element.</param>
  192. /// <param name="resultSelector">An asynchronous transform function to apply to each element of the intermediate sequence.</param>
  193. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function <paramref name="collectionSelector"/> on each element of the input sequence, awaiting the result, applying <paramref name="resultSelector"/> to each element of the intermediate sequences along with their corrasponding source element and awaiting the result.</returns>
  194. /// <exception cref="ArgumentNullException"><paramref name="source"/>, <paramref name="collectionSelector"/>, or <paramref name="resultSelector"/> is <see langword="null"/>.</exception>
  195. [GenerateAsyncOverload]
  196. private static IAsyncEnumerable<TResult> SelectManyAwaitCore<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, ValueTask<TResult>> resultSelector)
  197. {
  198. if (source == null)
  199. throw Error.ArgumentNull(nameof(source));
  200. if (collectionSelector == null)
  201. throw Error.ArgumentNull(nameof(collectionSelector));
  202. if (resultSelector == null)
  203. throw Error.ArgumentNull(nameof(resultSelector));
  204. return Core(source, collectionSelector, resultSelector);
  205. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, ValueTask<TResult>> resultSelector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  206. {
  207. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  208. {
  209. var inner = await collectionSelector(element).ConfigureAwait(false);
  210. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  211. {
  212. yield return await resultSelector(element, subElement).ConfigureAwait(false);
  213. }
  214. }
  215. }
  216. }
  217. #if !NO_DEEP_CANCELLATION
  218. [GenerateAsyncOverload]
  219. private static IAsyncEnumerable<TResult> SelectManyAwaitWithCancellationCore<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, CancellationToken, ValueTask<TResult>> resultSelector)
  220. {
  221. if (source == null)
  222. throw Error.ArgumentNull(nameof(source));
  223. if (collectionSelector == null)
  224. throw Error.ArgumentNull(nameof(collectionSelector));
  225. if (resultSelector == null)
  226. throw Error.ArgumentNull(nameof(resultSelector));
  227. return Core(source, collectionSelector, resultSelector);
  228. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, CancellationToken, ValueTask<TResult>> resultSelector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  229. {
  230. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  231. {
  232. var inner = await collectionSelector(element, cancellationToken).ConfigureAwait(false);
  233. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  234. {
  235. yield return await resultSelector(element, subElement, cancellationToken).ConfigureAwait(false);
  236. }
  237. }
  238. }
  239. }
  240. #endif
  241. /// <summary>
  242. /// Projects each element of an async-enumerable sequence to an async-enumerable sequence by incorporating the element's index, invokes the result selector for the source element and each of the corresponding inner sequence's elements, and merges the results into one async-enumerable sequence.
  243. /// </summary>
  244. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  245. /// <typeparam name="TCollection">The type of the elements in the projected intermediate sequences.</typeparam>
  246. /// <typeparam name="TResult">The type of the elements in the result sequence, obtained by using the selector to combine source sequence elements with their corresponding intermediate sequence elements.</typeparam>
  247. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  248. /// <param name="collectionSelector">A transform function to apply to each element; the second parameter of the function represents the index of the source element.</param>
  249. /// <param name="resultSelector">A transform function to apply to each element of the intermediate sequence; the second parameter of the function represents the index of the source element and the fourth parameter represents the index of the intermediate element.</param>
  250. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function collectionSelector on each element of the input sequence and then mapping each of those sequence elements and their corresponding source element to a result element.</returns>
  251. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="collectionSelector"/> or <paramref name="resultSelector"/> is null.</exception>
  252. public static IAsyncEnumerable<TResult> SelectMany<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector)
  253. {
  254. if (source == null)
  255. throw Error.ArgumentNull(nameof(source));
  256. if (collectionSelector == null)
  257. throw Error.ArgumentNull(nameof(collectionSelector));
  258. if (resultSelector == null)
  259. throw Error.ArgumentNull(nameof(resultSelector));
  260. return Core(source, collectionSelector, resultSelector);
  261. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  262. {
  263. var index = -1;
  264. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  265. {
  266. checked
  267. {
  268. index++;
  269. }
  270. var inner = collectionSelector(element, index);
  271. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  272. {
  273. yield return resultSelector(element, subElement);
  274. }
  275. }
  276. }
  277. }
  278. /// <summary>
  279. /// Projects each element of an async-enumerable sequence to an async-enumerable sequence by awaiting the result of a transform function that incorporates each element's index,
  280. /// invokes the result selector for the source element and each of the corrasponding inner-sequence's elements and awaits the result, and merges the results into one async-enumerable sequence.
  281. /// </summary>
  282. /// <typeparam name="TSource">The type of elements in the source sequence.</typeparam>
  283. /// <typeparam name="TCollection">The type of elements in the projected intermediate sequences.</typeparam>
  284. /// <typeparam name="TResult">The type of elements in the result sequence.</typeparam>
  285. /// <param name="source">An async-enumerable sequence of elements to project.</param>
  286. /// <param name="collectionSelector">An asynchronous transform function to apply to each source element; the second parameter represents the index of the element.</param>
  287. /// <param name="resultSelector">An asynchronous transform function to apply to each element of the intermediate sequence.</param>
  288. /// <returns>An async-enumerable sequence whose elements are the result of invoking the one-to-many transform function <paramref name="collectionSelector"/> on each element of the input sequence, awaiting the result, applying <paramref name="resultSelector"/> to each element of the intermediate sequences olong with their corrasponding source element and awaiting the result.</returns>
  289. /// <exception cref="ArgumentNullException"><paramref name="source"/>, <paramref name="collectionSelector"/>, or <paramref name="resultSelector"/> is <see langword="null"/>.</exception>
  290. [GenerateAsyncOverload]
  291. private static IAsyncEnumerable<TResult> SelectManyAwaitCore<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, ValueTask<TResult>> resultSelector)
  292. {
  293. if (source == null)
  294. throw Error.ArgumentNull(nameof(source));
  295. if (collectionSelector == null)
  296. throw Error.ArgumentNull(nameof(collectionSelector));
  297. if (resultSelector == null)
  298. throw Error.ArgumentNull(nameof(resultSelector));
  299. return Core(source, collectionSelector, resultSelector);
  300. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, int, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, ValueTask<TResult>> resultSelector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  301. {
  302. var index = -1;
  303. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  304. {
  305. checked
  306. {
  307. index++;
  308. }
  309. var inner = await collectionSelector(element, index).ConfigureAwait(false);
  310. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  311. {
  312. yield return await resultSelector(element, subElement).ConfigureAwait(false);
  313. }
  314. }
  315. }
  316. }
  317. #if !NO_DEEP_CANCELLATION
  318. [GenerateAsyncOverload]
  319. private static IAsyncEnumerable<TResult> SelectManyAwaitWithCancellationCore<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, CancellationToken, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, CancellationToken, ValueTask<TResult>> resultSelector)
  320. {
  321. if (source == null)
  322. throw Error.ArgumentNull(nameof(source));
  323. if (collectionSelector == null)
  324. throw Error.ArgumentNull(nameof(collectionSelector));
  325. if (resultSelector == null)
  326. throw Error.ArgumentNull(nameof(resultSelector));
  327. return Core(source, collectionSelector, resultSelector);
  328. static async IAsyncEnumerable<TResult> Core(IAsyncEnumerable<TSource> source, Func<TSource, int, CancellationToken, ValueTask<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, CancellationToken, ValueTask<TResult>> resultSelector, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
  329. {
  330. var index = -1;
  331. await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false))
  332. {
  333. checked
  334. {
  335. index++;
  336. }
  337. var inner = await collectionSelector(element, index, cancellationToken).ConfigureAwait(false);
  338. await foreach (var subElement in inner.WithCancellation(cancellationToken).ConfigureAwait(false))
  339. {
  340. yield return await resultSelector(element, subElement, cancellationToken).ConfigureAwait(false);
  341. }
  342. }
  343. }
  344. }
  345. #endif
  346. private sealed class SelectManyAsyncIterator<TSource, TResult> : AsyncIterator<TResult>, IAsyncIListProvider<TResult>
  347. {
  348. private const int State_Source = 1;
  349. private const int State_Result = 2;
  350. private readonly Func<TSource, IAsyncEnumerable<TResult>> _selector;
  351. private readonly IAsyncEnumerable<TSource> _source;
  352. private int _mode;
  353. private IAsyncEnumerator<TResult>? _resultEnumerator;
  354. private IAsyncEnumerator<TSource>? _sourceEnumerator;
  355. public SelectManyAsyncIterator(IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TResult>> selector)
  356. {
  357. _source = source;
  358. _selector = selector;
  359. }
  360. public override AsyncIteratorBase<TResult> Clone()
  361. {
  362. return new SelectManyAsyncIterator<TSource, TResult>(_source, _selector);
  363. }
  364. public override async ValueTask DisposeAsync()
  365. {
  366. if (_resultEnumerator != null)
  367. {
  368. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  369. _resultEnumerator = null;
  370. }
  371. if (_sourceEnumerator != null)
  372. {
  373. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  374. _sourceEnumerator = null;
  375. }
  376. await base.DisposeAsync().ConfigureAwait(false);
  377. }
  378. public ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken)
  379. {
  380. if (onlyIfCheap)
  381. {
  382. return new ValueTask<int>(-1);
  383. }
  384. return Core(cancellationToken);
  385. async ValueTask<int> Core(CancellationToken cancellationToken)
  386. {
  387. var count = 0;
  388. await foreach (var element in _source.WithCancellation(cancellationToken).ConfigureAwait(false))
  389. {
  390. checked
  391. {
  392. count += await _selector(element).CountAsync().ConfigureAwait(false);
  393. }
  394. }
  395. return count;
  396. }
  397. }
  398. public async ValueTask<TResult[]> ToArrayAsync(CancellationToken cancellationToken)
  399. {
  400. // REVIEW: Substitute for SparseArrayBuilder<T> logic once we have access to that.
  401. var list = await ToListAsync(cancellationToken).ConfigureAwait(false);
  402. return list.ToArray();
  403. }
  404. public async ValueTask<List<TResult>> ToListAsync(CancellationToken cancellationToken)
  405. {
  406. var list = new List<TResult>();
  407. await foreach (var element in _source.WithCancellation(cancellationToken).ConfigureAwait(false))
  408. {
  409. var items = _selector(element);
  410. await list.AddRangeAsync(items, cancellationToken).ConfigureAwait(false);
  411. }
  412. return list;
  413. }
  414. protected override async ValueTask<bool> MoveNextCore()
  415. {
  416. switch (_state)
  417. {
  418. case AsyncIteratorState.Allocated:
  419. _sourceEnumerator = _source.GetAsyncEnumerator(_cancellationToken);
  420. _mode = State_Source;
  421. _state = AsyncIteratorState.Iterating;
  422. goto case AsyncIteratorState.Iterating;
  423. case AsyncIteratorState.Iterating:
  424. switch (_mode)
  425. {
  426. case State_Source:
  427. if (await _sourceEnumerator!.MoveNextAsync().ConfigureAwait(false))
  428. {
  429. if (_resultEnumerator != null)
  430. {
  431. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  432. }
  433. var inner = _selector(_sourceEnumerator.Current);
  434. _resultEnumerator = inner.GetAsyncEnumerator(_cancellationToken);
  435. _mode = State_Result;
  436. goto case State_Result;
  437. }
  438. break;
  439. case State_Result:
  440. if (await _resultEnumerator!.MoveNextAsync().ConfigureAwait(false))
  441. {
  442. _current = _resultEnumerator.Current;
  443. return true;
  444. }
  445. _mode = State_Source;
  446. goto case State_Source; // loop
  447. }
  448. break;
  449. }
  450. await DisposeAsync().ConfigureAwait(false);
  451. return false;
  452. }
  453. }
  454. private sealed class SelectManyAsyncIteratorWithTask<TSource, TResult> : AsyncIterator<TResult>, IAsyncIListProvider<TResult>
  455. {
  456. private const int State_Source = 1;
  457. private const int State_Result = 2;
  458. private readonly Func<TSource, ValueTask<IAsyncEnumerable<TResult>>> _selector;
  459. private readonly IAsyncEnumerable<TSource> _source;
  460. private int _mode;
  461. private IAsyncEnumerator<TResult>? _resultEnumerator;
  462. private IAsyncEnumerator<TSource>? _sourceEnumerator;
  463. public SelectManyAsyncIteratorWithTask(IAsyncEnumerable<TSource> source, Func<TSource, ValueTask<IAsyncEnumerable<TResult>>> selector)
  464. {
  465. _source = source;
  466. _selector = selector;
  467. }
  468. public override AsyncIteratorBase<TResult> Clone()
  469. {
  470. return new SelectManyAsyncIteratorWithTask<TSource, TResult>(_source, _selector);
  471. }
  472. public override async ValueTask DisposeAsync()
  473. {
  474. if (_resultEnumerator != null)
  475. {
  476. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  477. _resultEnumerator = null;
  478. }
  479. if (_sourceEnumerator != null)
  480. {
  481. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  482. _sourceEnumerator = null;
  483. }
  484. await base.DisposeAsync().ConfigureAwait(false);
  485. }
  486. public ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken)
  487. {
  488. if (onlyIfCheap)
  489. {
  490. return new ValueTask<int>(-1);
  491. }
  492. return Core(cancellationToken);
  493. async ValueTask<int> Core(CancellationToken cancellationToken)
  494. {
  495. var count = 0;
  496. await foreach (var element in _source.WithCancellation(cancellationToken).ConfigureAwait(false))
  497. {
  498. var items = await _selector(element).ConfigureAwait(false);
  499. checked
  500. {
  501. count += await items.CountAsync().ConfigureAwait(false);
  502. }
  503. }
  504. return count;
  505. }
  506. }
  507. public async ValueTask<TResult[]> ToArrayAsync(CancellationToken cancellationToken)
  508. {
  509. // REVIEW: Substitute for SparseArrayBuilder<T> logic once we have access to that.
  510. var list = await ToListAsync(cancellationToken).ConfigureAwait(false);
  511. return list.ToArray();
  512. }
  513. public async ValueTask<List<TResult>> ToListAsync(CancellationToken cancellationToken)
  514. {
  515. var list = new List<TResult>();
  516. await foreach (var element in _source.WithCancellation(cancellationToken).ConfigureAwait(false))
  517. {
  518. var items = await _selector(element).ConfigureAwait(false);
  519. await list.AddRangeAsync(items, cancellationToken).ConfigureAwait(false);
  520. }
  521. return list;
  522. }
  523. protected override async ValueTask<bool> MoveNextCore()
  524. {
  525. switch (_state)
  526. {
  527. case AsyncIteratorState.Allocated:
  528. _sourceEnumerator = _source.GetAsyncEnumerator(_cancellationToken);
  529. _mode = State_Source;
  530. _state = AsyncIteratorState.Iterating;
  531. goto case AsyncIteratorState.Iterating;
  532. case AsyncIteratorState.Iterating:
  533. switch (_mode)
  534. {
  535. case State_Source:
  536. if (await _sourceEnumerator!.MoveNextAsync().ConfigureAwait(false))
  537. {
  538. if (_resultEnumerator != null)
  539. {
  540. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  541. }
  542. var inner = await _selector(_sourceEnumerator.Current).ConfigureAwait(false);
  543. _resultEnumerator = inner.GetAsyncEnumerator(_cancellationToken);
  544. _mode = State_Result;
  545. goto case State_Result;
  546. }
  547. break;
  548. case State_Result:
  549. if (await _resultEnumerator!.MoveNextAsync().ConfigureAwait(false))
  550. {
  551. _current = _resultEnumerator.Current;
  552. return true;
  553. }
  554. _mode = State_Source;
  555. goto case State_Source; // loop
  556. }
  557. break;
  558. }
  559. await DisposeAsync().ConfigureAwait(false);
  560. return false;
  561. }
  562. }
  563. #if !NO_DEEP_CANCELLATION
  564. private sealed class SelectManyAsyncIteratorWithTaskAndCancellation<TSource, TResult> : AsyncIterator<TResult>, IAsyncIListProvider<TResult>
  565. {
  566. private const int State_Source = 1;
  567. private const int State_Result = 2;
  568. private readonly Func<TSource, CancellationToken, ValueTask<IAsyncEnumerable<TResult>>> _selector;
  569. private readonly IAsyncEnumerable<TSource> _source;
  570. private int _mode;
  571. private IAsyncEnumerator<TResult>? _resultEnumerator;
  572. private IAsyncEnumerator<TSource>? _sourceEnumerator;
  573. public SelectManyAsyncIteratorWithTaskAndCancellation(IAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, ValueTask<IAsyncEnumerable<TResult>>> selector)
  574. {
  575. _source = source;
  576. _selector = selector;
  577. }
  578. public override AsyncIteratorBase<TResult> Clone()
  579. {
  580. return new SelectManyAsyncIteratorWithTaskAndCancellation<TSource, TResult>(_source, _selector);
  581. }
  582. public override async ValueTask DisposeAsync()
  583. {
  584. if (_resultEnumerator != null)
  585. {
  586. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  587. _resultEnumerator = null;
  588. }
  589. if (_sourceEnumerator != null)
  590. {
  591. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  592. _sourceEnumerator = null;
  593. }
  594. await base.DisposeAsync().ConfigureAwait(false);
  595. }
  596. public ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken)
  597. {
  598. if (onlyIfCheap)
  599. {
  600. return new ValueTask<int>(-1);
  601. }
  602. return Core(cancellationToken);
  603. async ValueTask<int> Core(CancellationToken cancellationToken)
  604. {
  605. var count = 0;
  606. await foreach (var element in _source.WithCancellation(cancellationToken).ConfigureAwait(false))
  607. {
  608. var items = await _selector(element, cancellationToken).ConfigureAwait(false);
  609. checked
  610. {
  611. count += await items.CountAsync().ConfigureAwait(false);
  612. }
  613. }
  614. return count;
  615. }
  616. }
  617. public async ValueTask<TResult[]> ToArrayAsync(CancellationToken cancellationToken)
  618. {
  619. // REVIEW: Substitute for SparseArrayBuilder<T> logic once we have access to that.
  620. var list = await ToListAsync(cancellationToken).ConfigureAwait(false);
  621. return list.ToArray();
  622. }
  623. public async ValueTask<List<TResult>> ToListAsync(CancellationToken cancellationToken)
  624. {
  625. var list = new List<TResult>();
  626. await foreach (var element in _source.WithCancellation(cancellationToken).ConfigureAwait(false))
  627. {
  628. var items = await _selector(element, cancellationToken).ConfigureAwait(false);
  629. await list.AddRangeAsync(items, cancellationToken).ConfigureAwait(false);
  630. }
  631. return list;
  632. }
  633. protected override async ValueTask<bool> MoveNextCore()
  634. {
  635. switch (_state)
  636. {
  637. case AsyncIteratorState.Allocated:
  638. _sourceEnumerator = _source.GetAsyncEnumerator(_cancellationToken);
  639. _mode = State_Source;
  640. _state = AsyncIteratorState.Iterating;
  641. goto case AsyncIteratorState.Iterating;
  642. case AsyncIteratorState.Iterating:
  643. switch (_mode)
  644. {
  645. case State_Source:
  646. if (await _sourceEnumerator!.MoveNextAsync().ConfigureAwait(false))
  647. {
  648. if (_resultEnumerator != null)
  649. {
  650. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  651. }
  652. var inner = await _selector(_sourceEnumerator.Current, _cancellationToken).ConfigureAwait(false);
  653. _resultEnumerator = inner.GetAsyncEnumerator(_cancellationToken);
  654. _mode = State_Result;
  655. goto case State_Result;
  656. }
  657. break;
  658. case State_Result:
  659. if (await _resultEnumerator!.MoveNextAsync().ConfigureAwait(false))
  660. {
  661. _current = _resultEnumerator.Current;
  662. return true;
  663. }
  664. _mode = State_Source;
  665. goto case State_Source; // loop
  666. }
  667. break;
  668. }
  669. await DisposeAsync().ConfigureAwait(false);
  670. return false;
  671. }
  672. }
  673. #endif
  674. }
  675. }