SelectMany.cs 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888
  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.Diagnostics;
  6. using System.Threading;
  7. using System.Threading.Tasks;
  8. namespace System.Linq
  9. {
  10. public static partial class AsyncEnumerable
  11. {
  12. public static IAsyncEnumerable<TResult> SelectMany<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TResult>> selector)
  13. {
  14. if (source == null)
  15. throw new ArgumentNullException(nameof(source));
  16. if (selector == null)
  17. throw new ArgumentNullException(nameof(selector));
  18. return new SelectManyAsyncIterator<TSource, TResult>(source, selector);
  19. }
  20. public static IAsyncEnumerable<TResult> SelectMany<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<IAsyncEnumerable<TResult>>> selector)
  21. {
  22. if (source == null)
  23. throw new ArgumentNullException(nameof(source));
  24. if (selector == null)
  25. throw new ArgumentNullException(nameof(selector));
  26. return new SelectManyAsyncIteratorWithTask<TSource, TResult>(source, selector);
  27. }
  28. public static IAsyncEnumerable<TResult> SelectMany<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TResult>> selector)
  29. {
  30. if (source == null)
  31. throw new ArgumentNullException(nameof(source));
  32. if (selector == null)
  33. throw new ArgumentNullException(nameof(selector));
  34. return new SelectManyWithIndexAsyncIterator<TSource, TResult>(source, selector);
  35. }
  36. public static IAsyncEnumerable<TResult> SelectMany<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, Task<IAsyncEnumerable<TResult>>> selector)
  37. {
  38. if (source == null)
  39. throw new ArgumentNullException(nameof(source));
  40. if (selector == null)
  41. throw new ArgumentNullException(nameof(selector));
  42. return new SelectManyWithIndexAsyncIteratorWithTask<TSource, TResult>(source, selector);
  43. }
  44. public static IAsyncEnumerable<TResult> SelectMany<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TCollection>> selector, Func<TSource, TCollection, TResult> resultSelector)
  45. {
  46. if (source == null)
  47. throw new ArgumentNullException(nameof(source));
  48. if (selector == null)
  49. throw new ArgumentNullException(nameof(selector));
  50. if (resultSelector == null)
  51. throw new ArgumentNullException(nameof(resultSelector));
  52. return new SelectManyAsyncIterator<TSource, TCollection, TResult>(source, selector, resultSelector);
  53. }
  54. public static IAsyncEnumerable<TResult> SelectMany<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<IAsyncEnumerable<TCollection>>> selector, Func<TSource, TCollection, Task<TResult>> resultSelector)
  55. {
  56. if (source == null)
  57. throw new ArgumentNullException(nameof(source));
  58. if (selector == null)
  59. throw new ArgumentNullException(nameof(selector));
  60. if (resultSelector == null)
  61. throw new ArgumentNullException(nameof(resultSelector));
  62. return new SelectManyAsyncIteratorWithTask<TSource, TCollection, TResult>(source, selector, resultSelector);
  63. }
  64. public static IAsyncEnumerable<TResult> SelectMany<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TCollection>> selector, Func<TSource, TCollection, TResult> resultSelector)
  65. {
  66. if (source == null)
  67. throw new ArgumentNullException(nameof(source));
  68. if (selector == null)
  69. throw new ArgumentNullException(nameof(selector));
  70. if (resultSelector == null)
  71. throw new ArgumentNullException(nameof(resultSelector));
  72. return new SelectManyWithIndexAsyncIterator<TSource, TCollection, TResult>(source, selector, resultSelector);
  73. }
  74. public static IAsyncEnumerable<TResult> SelectMany<TSource, TCollection, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, int, Task<IAsyncEnumerable<TCollection>>> selector, Func<TSource, TCollection, Task<TResult>> resultSelector)
  75. {
  76. if (source == null)
  77. throw new ArgumentNullException(nameof(source));
  78. if (selector == null)
  79. throw new ArgumentNullException(nameof(selector));
  80. if (resultSelector == null)
  81. throw new ArgumentNullException(nameof(resultSelector));
  82. return new SelectManyWithIndexAsyncIteratorWithTask<TSource, TCollection, TResult>(source, selector, resultSelector);
  83. }
  84. private sealed class SelectManyAsyncIterator<TSource, TResult> : AsyncIterator<TResult>
  85. {
  86. private const int State_Source = 1;
  87. private const int State_Result = 2;
  88. private readonly Func<TSource, IAsyncEnumerable<TResult>> _selector;
  89. private readonly IAsyncEnumerable<TSource> _source;
  90. private int _mode;
  91. private IAsyncEnumerator<TResult> _resultEnumerator;
  92. private IAsyncEnumerator<TSource> _sourceEnumerator;
  93. public SelectManyAsyncIterator(IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TResult>> selector)
  94. {
  95. Debug.Assert(source != null);
  96. Debug.Assert(selector != null);
  97. _source = source;
  98. _selector = selector;
  99. }
  100. public override AsyncIterator<TResult> Clone()
  101. {
  102. return new SelectManyAsyncIterator<TSource, TResult>(_source, _selector);
  103. }
  104. public override async ValueTask DisposeAsync()
  105. {
  106. if (_sourceEnumerator != null)
  107. {
  108. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  109. _sourceEnumerator = null;
  110. }
  111. if (_resultEnumerator != null)
  112. {
  113. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  114. _resultEnumerator = null;
  115. }
  116. await base.DisposeAsync().ConfigureAwait(false);
  117. }
  118. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  119. {
  120. switch (state)
  121. {
  122. case AsyncIteratorState.Allocated:
  123. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  124. _mode = State_Source;
  125. state = AsyncIteratorState.Iterating;
  126. goto case AsyncIteratorState.Iterating;
  127. case AsyncIteratorState.Iterating:
  128. switch (_mode)
  129. {
  130. case State_Source:
  131. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  132. {
  133. if (_resultEnumerator != null)
  134. {
  135. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  136. }
  137. var inner = _selector(_sourceEnumerator.Current);
  138. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  139. _mode = State_Result;
  140. goto case State_Result;
  141. }
  142. break;
  143. case State_Result:
  144. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  145. {
  146. current = _resultEnumerator.Current;
  147. return true;
  148. }
  149. _mode = State_Source;
  150. goto case State_Source; // loop
  151. }
  152. break;
  153. }
  154. await DisposeAsync().ConfigureAwait(false);
  155. return false;
  156. }
  157. }
  158. private sealed class SelectManyAsyncIteratorWithTask<TSource, TResult> : AsyncIterator<TResult>
  159. {
  160. private const int State_Source = 1;
  161. private const int State_Result = 2;
  162. private readonly Func<TSource, Task<IAsyncEnumerable<TResult>>> _selector;
  163. private readonly IAsyncEnumerable<TSource> _source;
  164. private int _mode;
  165. private IAsyncEnumerator<TResult> _resultEnumerator;
  166. private IAsyncEnumerator<TSource> _sourceEnumerator;
  167. public SelectManyAsyncIteratorWithTask(IAsyncEnumerable<TSource> source, Func<TSource, Task<IAsyncEnumerable<TResult>>> selector)
  168. {
  169. Debug.Assert(source != null);
  170. Debug.Assert(selector != null);
  171. _source = source;
  172. _selector = selector;
  173. }
  174. public override AsyncIterator<TResult> Clone()
  175. {
  176. return new SelectManyAsyncIteratorWithTask<TSource, TResult>(_source, _selector);
  177. }
  178. public override async ValueTask DisposeAsync()
  179. {
  180. if (_sourceEnumerator != null)
  181. {
  182. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  183. _sourceEnumerator = null;
  184. }
  185. if (_resultEnumerator != null)
  186. {
  187. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  188. _resultEnumerator = null;
  189. }
  190. await base.DisposeAsync().ConfigureAwait(false);
  191. }
  192. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  193. {
  194. switch (state)
  195. {
  196. case AsyncIteratorState.Allocated:
  197. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  198. _mode = State_Source;
  199. state = AsyncIteratorState.Iterating;
  200. goto case AsyncIteratorState.Iterating;
  201. case AsyncIteratorState.Iterating:
  202. switch (_mode)
  203. {
  204. case State_Source:
  205. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  206. {
  207. if (_resultEnumerator != null)
  208. {
  209. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  210. }
  211. var inner = await _selector(_sourceEnumerator.Current).ConfigureAwait(false);
  212. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  213. _mode = State_Result;
  214. goto case State_Result;
  215. }
  216. break;
  217. case State_Result:
  218. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  219. {
  220. current = _resultEnumerator.Current;
  221. return true;
  222. }
  223. _mode = State_Source;
  224. goto case State_Source; // loop
  225. }
  226. break;
  227. }
  228. await DisposeAsync().ConfigureAwait(false);
  229. return false;
  230. }
  231. }
  232. private sealed class SelectManyAsyncIterator<TSource, TCollection, TResult> : AsyncIterator<TResult>
  233. {
  234. private const int State_Source = 1;
  235. private const int State_Result = 2;
  236. private readonly Func<TSource, IAsyncEnumerable<TCollection>> _collectionSelector;
  237. private readonly Func<TSource, TCollection, TResult> _resultSelector;
  238. private readonly IAsyncEnumerable<TSource> _source;
  239. private TSource _currentSource;
  240. private int _mode;
  241. private IAsyncEnumerator<TCollection> _resultEnumerator;
  242. private IAsyncEnumerator<TSource> _sourceEnumerator;
  243. public SelectManyAsyncIterator(IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector)
  244. {
  245. Debug.Assert(source != null);
  246. Debug.Assert(collectionSelector != null);
  247. Debug.Assert(resultSelector != null);
  248. _source = source;
  249. _collectionSelector = collectionSelector;
  250. _resultSelector = resultSelector;
  251. }
  252. public override AsyncIterator<TResult> Clone()
  253. {
  254. return new SelectManyAsyncIterator<TSource, TCollection, TResult>(_source, _collectionSelector, _resultSelector);
  255. }
  256. public override async ValueTask DisposeAsync()
  257. {
  258. if (_sourceEnumerator != null)
  259. {
  260. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  261. _sourceEnumerator = null;
  262. }
  263. if (_resultEnumerator != null)
  264. {
  265. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  266. _resultEnumerator = null;
  267. }
  268. _currentSource = default;
  269. await base.DisposeAsync().ConfigureAwait(false);
  270. }
  271. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  272. {
  273. switch (state)
  274. {
  275. case AsyncIteratorState.Allocated:
  276. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  277. _mode = State_Source;
  278. state = AsyncIteratorState.Iterating;
  279. goto case AsyncIteratorState.Iterating;
  280. case AsyncIteratorState.Iterating:
  281. switch (_mode)
  282. {
  283. case State_Source:
  284. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  285. {
  286. if (_resultEnumerator != null)
  287. {
  288. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  289. }
  290. _currentSource = _sourceEnumerator.Current;
  291. var inner = _collectionSelector(_currentSource);
  292. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  293. _mode = State_Result;
  294. goto case State_Result;
  295. }
  296. break;
  297. case State_Result:
  298. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  299. {
  300. current = _resultSelector(_currentSource, _resultEnumerator.Current);
  301. return true;
  302. }
  303. _mode = State_Source;
  304. goto case State_Source; // loop
  305. }
  306. break;
  307. }
  308. await DisposeAsync().ConfigureAwait(false);
  309. return false;
  310. }
  311. }
  312. private sealed class SelectManyAsyncIteratorWithTask<TSource, TCollection, TResult> : AsyncIterator<TResult>
  313. {
  314. private const int State_Source = 1;
  315. private const int State_Result = 2;
  316. private readonly Func<TSource, Task<IAsyncEnumerable<TCollection>>> _collectionSelector;
  317. private readonly Func<TSource, TCollection, Task<TResult>> _resultSelector;
  318. private readonly IAsyncEnumerable<TSource> _source;
  319. private TSource _currentSource;
  320. private int _mode;
  321. private IAsyncEnumerator<TCollection> _resultEnumerator;
  322. private IAsyncEnumerator<TSource> _sourceEnumerator;
  323. public SelectManyAsyncIteratorWithTask(IAsyncEnumerable<TSource> source, Func<TSource, Task<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, Task<TResult>> resultSelector)
  324. {
  325. Debug.Assert(source != null);
  326. Debug.Assert(collectionSelector != null);
  327. Debug.Assert(resultSelector != null);
  328. _source = source;
  329. _collectionSelector = collectionSelector;
  330. _resultSelector = resultSelector;
  331. }
  332. public override AsyncIterator<TResult> Clone()
  333. {
  334. return new SelectManyAsyncIteratorWithTask<TSource, TCollection, TResult>(_source, _collectionSelector, _resultSelector);
  335. }
  336. public override async ValueTask DisposeAsync()
  337. {
  338. if (_sourceEnumerator != null)
  339. {
  340. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  341. _sourceEnumerator = null;
  342. }
  343. if (_resultEnumerator != null)
  344. {
  345. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  346. _resultEnumerator = null;
  347. }
  348. _currentSource = default;
  349. await base.DisposeAsync().ConfigureAwait(false);
  350. }
  351. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  352. {
  353. switch (state)
  354. {
  355. case AsyncIteratorState.Allocated:
  356. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  357. _mode = State_Source;
  358. state = AsyncIteratorState.Iterating;
  359. goto case AsyncIteratorState.Iterating;
  360. case AsyncIteratorState.Iterating:
  361. switch (_mode)
  362. {
  363. case State_Source:
  364. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  365. {
  366. if (_resultEnumerator != null)
  367. {
  368. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  369. }
  370. _currentSource = _sourceEnumerator.Current;
  371. var inner = await _collectionSelector(_currentSource).ConfigureAwait(false);
  372. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  373. _mode = State_Result;
  374. goto case State_Result;
  375. }
  376. break;
  377. case State_Result:
  378. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  379. {
  380. current = await _resultSelector(_currentSource, _resultEnumerator.Current).ConfigureAwait(false);
  381. return true;
  382. }
  383. _mode = State_Source;
  384. goto case State_Source; // loop
  385. }
  386. break;
  387. }
  388. await DisposeAsync().ConfigureAwait(false);
  389. return false;
  390. }
  391. }
  392. private sealed class SelectManyWithIndexAsyncIterator<TSource, TCollection, TResult> : AsyncIterator<TResult>
  393. {
  394. private const int State_Source = 1;
  395. private const int State_Result = 2;
  396. private readonly Func<TSource, int, IAsyncEnumerable<TCollection>> _collectionSelector;
  397. private readonly Func<TSource, TCollection, TResult> _resultSelector;
  398. private readonly IAsyncEnumerable<TSource> _source;
  399. private TSource _currentSource;
  400. private int _index;
  401. private int _mode;
  402. private IAsyncEnumerator<TCollection> _resultEnumerator;
  403. private IAsyncEnumerator<TSource> _sourceEnumerator;
  404. public SelectManyWithIndexAsyncIterator(IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector)
  405. {
  406. Debug.Assert(source != null);
  407. Debug.Assert(collectionSelector != null);
  408. Debug.Assert(resultSelector != null);
  409. _source = source;
  410. _collectionSelector = collectionSelector;
  411. _resultSelector = resultSelector;
  412. }
  413. public override AsyncIterator<TResult> Clone()
  414. {
  415. return new SelectManyWithIndexAsyncIterator<TSource, TCollection, TResult>(_source, _collectionSelector, _resultSelector);
  416. }
  417. public override async ValueTask DisposeAsync()
  418. {
  419. if (_sourceEnumerator != null)
  420. {
  421. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  422. _sourceEnumerator = null;
  423. }
  424. if (_resultEnumerator != null)
  425. {
  426. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  427. _resultEnumerator = null;
  428. }
  429. _currentSource = default;
  430. await base.DisposeAsync().ConfigureAwait(false);
  431. }
  432. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  433. {
  434. switch (state)
  435. {
  436. case AsyncIteratorState.Allocated:
  437. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  438. _index = -1;
  439. _mode = State_Source;
  440. state = AsyncIteratorState.Iterating;
  441. goto case AsyncIteratorState.Iterating;
  442. case AsyncIteratorState.Iterating:
  443. switch (_mode)
  444. {
  445. case State_Source:
  446. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  447. {
  448. if (_resultEnumerator != null)
  449. {
  450. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  451. }
  452. _currentSource = _sourceEnumerator.Current;
  453. checked
  454. {
  455. _index++;
  456. }
  457. var inner = _collectionSelector(_currentSource, _index);
  458. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  459. _mode = State_Result;
  460. goto case State_Result;
  461. }
  462. break;
  463. case State_Result:
  464. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  465. {
  466. current = _resultSelector(_currentSource, _resultEnumerator.Current);
  467. return true;
  468. }
  469. _mode = State_Source;
  470. goto case State_Source; // loop
  471. }
  472. break;
  473. }
  474. await DisposeAsync().ConfigureAwait(false);
  475. return false;
  476. }
  477. }
  478. private sealed class SelectManyWithIndexAsyncIteratorWithTask<TSource, TCollection, TResult> : AsyncIterator<TResult>
  479. {
  480. private const int State_Source = 1;
  481. private const int State_Result = 2;
  482. private readonly Func<TSource, int, Task<IAsyncEnumerable<TCollection>>> _collectionSelector;
  483. private readonly Func<TSource, TCollection, Task<TResult>> _resultSelector;
  484. private readonly IAsyncEnumerable<TSource> _source;
  485. private TSource _currentSource;
  486. private int _index;
  487. private int _mode;
  488. private IAsyncEnumerator<TCollection> _resultEnumerator;
  489. private IAsyncEnumerator<TSource> _sourceEnumerator;
  490. public SelectManyWithIndexAsyncIteratorWithTask(IAsyncEnumerable<TSource> source, Func<TSource, int, Task<IAsyncEnumerable<TCollection>>> collectionSelector, Func<TSource, TCollection, Task<TResult>> resultSelector)
  491. {
  492. Debug.Assert(source != null);
  493. Debug.Assert(collectionSelector != null);
  494. Debug.Assert(resultSelector != null);
  495. _source = source;
  496. _collectionSelector = collectionSelector;
  497. _resultSelector = resultSelector;
  498. }
  499. public override AsyncIterator<TResult> Clone()
  500. {
  501. return new SelectManyWithIndexAsyncIteratorWithTask<TSource, TCollection, TResult>(_source, _collectionSelector, _resultSelector);
  502. }
  503. public override async ValueTask DisposeAsync()
  504. {
  505. if (_sourceEnumerator != null)
  506. {
  507. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  508. _sourceEnumerator = null;
  509. }
  510. if (_resultEnumerator != null)
  511. {
  512. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  513. _resultEnumerator = null;
  514. }
  515. _currentSource = default;
  516. await base.DisposeAsync().ConfigureAwait(false);
  517. }
  518. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  519. {
  520. switch (state)
  521. {
  522. case AsyncIteratorState.Allocated:
  523. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  524. _index = -1;
  525. _mode = State_Source;
  526. state = AsyncIteratorState.Iterating;
  527. goto case AsyncIteratorState.Iterating;
  528. case AsyncIteratorState.Iterating:
  529. switch (_mode)
  530. {
  531. case State_Source:
  532. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  533. {
  534. if (_resultEnumerator != null)
  535. {
  536. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  537. }
  538. _currentSource = _sourceEnumerator.Current;
  539. checked
  540. {
  541. _index++;
  542. }
  543. var inner = await _collectionSelector(_currentSource, _index).ConfigureAwait(false);
  544. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  545. _mode = State_Result;
  546. goto case State_Result;
  547. }
  548. break;
  549. case State_Result:
  550. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  551. {
  552. current = await _resultSelector(_currentSource, _resultEnumerator.Current).ConfigureAwait(false);
  553. return true;
  554. }
  555. _mode = State_Source;
  556. goto case State_Source; // loop
  557. }
  558. break;
  559. }
  560. await DisposeAsync().ConfigureAwait(false);
  561. return false;
  562. }
  563. }
  564. private sealed class SelectManyWithIndexAsyncIterator<TSource, TResult> : AsyncIterator<TResult>
  565. {
  566. private const int State_Source = 1;
  567. private const int State_Result = 2;
  568. private readonly Func<TSource, int, IAsyncEnumerable<TResult>> _selector;
  569. private readonly IAsyncEnumerable<TSource> _source;
  570. private int _index;
  571. private int _mode;
  572. private IAsyncEnumerator<TResult> _resultEnumerator;
  573. private IAsyncEnumerator<TSource> _sourceEnumerator;
  574. public SelectManyWithIndexAsyncIterator(IAsyncEnumerable<TSource> source, Func<TSource, int, IAsyncEnumerable<TResult>> selector)
  575. {
  576. Debug.Assert(source != null);
  577. Debug.Assert(selector != null);
  578. _source = source;
  579. _selector = selector;
  580. }
  581. public override AsyncIterator<TResult> Clone()
  582. {
  583. return new SelectManyWithIndexAsyncIterator<TSource, TResult>(_source, _selector);
  584. }
  585. public override async ValueTask DisposeAsync()
  586. {
  587. if (_sourceEnumerator != null)
  588. {
  589. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  590. _sourceEnumerator = null;
  591. }
  592. if (_resultEnumerator != null)
  593. {
  594. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  595. _resultEnumerator = null;
  596. }
  597. await base.DisposeAsync().ConfigureAwait(false);
  598. }
  599. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  600. {
  601. switch (state)
  602. {
  603. case AsyncIteratorState.Allocated:
  604. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  605. _index = -1;
  606. _mode = State_Source;
  607. state = AsyncIteratorState.Iterating;
  608. goto case AsyncIteratorState.Iterating;
  609. case AsyncIteratorState.Iterating:
  610. switch (_mode)
  611. {
  612. case State_Source:
  613. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  614. {
  615. if (_resultEnumerator != null)
  616. {
  617. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  618. }
  619. checked
  620. {
  621. _index++;
  622. }
  623. var inner = _selector(_sourceEnumerator.Current, _index);
  624. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  625. _mode = State_Result;
  626. goto case State_Result;
  627. }
  628. break;
  629. case State_Result:
  630. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  631. {
  632. current = _resultEnumerator.Current;
  633. return true;
  634. }
  635. _mode = State_Source;
  636. goto case State_Source; // loop
  637. }
  638. break;
  639. }
  640. await DisposeAsync().ConfigureAwait(false);
  641. return false;
  642. }
  643. }
  644. private sealed class SelectManyWithIndexAsyncIteratorWithTask<TSource, TResult> : AsyncIterator<TResult>
  645. {
  646. private const int State_Source = 1;
  647. private const int State_Result = 2;
  648. private readonly Func<TSource, int, Task<IAsyncEnumerable<TResult>>> _selector;
  649. private readonly IAsyncEnumerable<TSource> _source;
  650. private int _index;
  651. private int _mode;
  652. private IAsyncEnumerator<TResult> _resultEnumerator;
  653. private IAsyncEnumerator<TSource> _sourceEnumerator;
  654. public SelectManyWithIndexAsyncIteratorWithTask(IAsyncEnumerable<TSource> source, Func<TSource, int, Task<IAsyncEnumerable<TResult>>> selector)
  655. {
  656. Debug.Assert(source != null);
  657. Debug.Assert(selector != null);
  658. _source = source;
  659. _selector = selector;
  660. }
  661. public override AsyncIterator<TResult> Clone()
  662. {
  663. return new SelectManyWithIndexAsyncIteratorWithTask<TSource, TResult>(_source, _selector);
  664. }
  665. public override async ValueTask DisposeAsync()
  666. {
  667. if (_sourceEnumerator != null)
  668. {
  669. await _sourceEnumerator.DisposeAsync().ConfigureAwait(false);
  670. _sourceEnumerator = null;
  671. }
  672. if (_resultEnumerator != null)
  673. {
  674. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  675. _resultEnumerator = null;
  676. }
  677. await base.DisposeAsync().ConfigureAwait(false);
  678. }
  679. protected override async ValueTask<bool> MoveNextCore(CancellationToken cancellationToken)
  680. {
  681. switch (state)
  682. {
  683. case AsyncIteratorState.Allocated:
  684. _sourceEnumerator = _source.GetAsyncEnumerator(cancellationToken);
  685. _index = -1;
  686. _mode = State_Source;
  687. state = AsyncIteratorState.Iterating;
  688. goto case AsyncIteratorState.Iterating;
  689. case AsyncIteratorState.Iterating:
  690. switch (_mode)
  691. {
  692. case State_Source:
  693. if (await _sourceEnumerator.MoveNextAsync().ConfigureAwait(false))
  694. {
  695. if (_resultEnumerator != null)
  696. {
  697. await _resultEnumerator.DisposeAsync().ConfigureAwait(false);
  698. }
  699. checked
  700. {
  701. _index++;
  702. }
  703. var inner = await _selector(_sourceEnumerator.Current, _index).ConfigureAwait(false);
  704. _resultEnumerator = inner.GetAsyncEnumerator(cancellationToken);
  705. _mode = State_Result;
  706. goto case State_Result;
  707. }
  708. break;
  709. case State_Result:
  710. if (await _resultEnumerator.MoveNextAsync().ConfigureAwait(false))
  711. {
  712. current = _resultEnumerator.Current;
  713. return true;
  714. }
  715. _mode = State_Source;
  716. goto case State_Source; // loop
  717. }
  718. break;
  719. }
  720. await DisposeAsync().ConfigureAwait(false);
  721. return false;
  722. }
  723. }
  724. }
  725. }