GroupBy.cs 44 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923
  1. using Cysharp.Threading.Tasks.Internal;
  2. using System;
  3. using System.Collections.Generic;
  4. using System.Linq;
  5. using System.Threading;
  6. namespace Cysharp.Threading.Tasks.Linq
  7. {
  8. public static partial class UniTaskAsyncEnumerable
  9. {
  10. // Ix-Async returns IGrouping but it is competely waste, use standard IGrouping.
  11. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TSource>> GroupBy<TSource, TKey>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector)
  12. {
  13. Error.ThrowArgumentNullException(source, nameof(source));
  14. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  15. return new GroupBy<TSource, TKey, TSource>(source, keySelector, x => x, EqualityComparer<TKey>.Default);
  16. }
  17. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TSource>> GroupBy<TSource, TKey>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> comparer)
  18. {
  19. Error.ThrowArgumentNullException(source, nameof(source));
  20. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  21. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  22. return new GroupBy<TSource, TKey, TSource>(source, keySelector, x => x, comparer);
  23. }
  24. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector)
  25. {
  26. Error.ThrowArgumentNullException(source, nameof(source));
  27. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  28. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  29. return new GroupBy<TSource, TKey, TElement>(source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
  30. }
  31. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, IEqualityComparer<TKey> comparer)
  32. {
  33. Error.ThrowArgumentNullException(source, nameof(source));
  34. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  35. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  36. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  37. return new GroupBy<TSource, TKey, TElement>(source, keySelector, elementSelector, comparer);
  38. }
  39. public static IUniTaskAsyncEnumerable<TResult> GroupBy<TSource, TKey, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TKey, IEnumerable<TSource>, TResult> resultSelector)
  40. {
  41. Error.ThrowArgumentNullException(source, nameof(source));
  42. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  43. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  44. return new GroupBy<TSource, TKey, TSource, TResult>(source, keySelector, x => x, resultSelector, EqualityComparer<TKey>.Default);
  45. }
  46. public static IUniTaskAsyncEnumerable<TResult> GroupBy<TSource, TKey, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TKey, IEnumerable<TSource>, TResult> resultSelector, IEqualityComparer<TKey> comparer)
  47. {
  48. Error.ThrowArgumentNullException(source, nameof(source));
  49. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  50. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  51. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  52. return new GroupBy<TSource, TKey, TSource, TResult>(source, keySelector, x => x, resultSelector, comparer);
  53. }
  54. public static IUniTaskAsyncEnumerable<TResult> GroupBy<TSource, TKey, TElement, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, Func<TKey, IEnumerable<TElement>, TResult> resultSelector)
  55. {
  56. Error.ThrowArgumentNullException(source, nameof(source));
  57. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  58. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  59. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  60. return new GroupBy<TSource, TKey, TElement, TResult>(source, keySelector, elementSelector, resultSelector, EqualityComparer<TKey>.Default);
  61. }
  62. public static IUniTaskAsyncEnumerable<TResult> GroupBy<TSource, TKey, TElement, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, Func<TKey, IEnumerable<TElement>, TResult> resultSelector, IEqualityComparer<TKey> comparer)
  63. {
  64. Error.ThrowArgumentNullException(source, nameof(source));
  65. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  66. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  67. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  68. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  69. return new GroupBy<TSource, TKey, TElement, TResult>(source, keySelector, elementSelector, resultSelector, comparer);
  70. }
  71. // await
  72. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TSource>> GroupByAwait<TSource, TKey>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector)
  73. {
  74. Error.ThrowArgumentNullException(source, nameof(source));
  75. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  76. return new GroupByAwait<TSource, TKey, TSource>(source, keySelector, x => UniTask.FromResult(x), EqualityComparer<TKey>.Default);
  77. }
  78. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TSource>> GroupByAwait<TSource, TKey>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, IEqualityComparer<TKey> comparer)
  79. {
  80. Error.ThrowArgumentNullException(source, nameof(source));
  81. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  82. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  83. return new GroupByAwait<TSource, TKey, TSource>(source, keySelector, x => UniTask.FromResult(x), comparer);
  84. }
  85. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>> GroupByAwait<TSource, TKey, TElement>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector)
  86. {
  87. Error.ThrowArgumentNullException(source, nameof(source));
  88. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  89. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  90. return new GroupByAwait<TSource, TKey, TElement>(source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
  91. }
  92. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>> GroupByAwait<TSource, TKey, TElement>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, IEqualityComparer<TKey> comparer)
  93. {
  94. Error.ThrowArgumentNullException(source, nameof(source));
  95. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  96. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  97. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  98. return new GroupByAwait<TSource, TKey, TElement>(source, keySelector, elementSelector, comparer);
  99. }
  100. public static IUniTaskAsyncEnumerable<TResult> GroupByAwait<TSource, TKey, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TKey, IEnumerable<TSource>, UniTask<TResult>> resultSelector)
  101. {
  102. Error.ThrowArgumentNullException(source, nameof(source));
  103. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  104. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  105. return new GroupByAwait<TSource, TKey, TSource, TResult>(source, keySelector, x => UniTask.FromResult(x), resultSelector, EqualityComparer<TKey>.Default);
  106. }
  107. public static IUniTaskAsyncEnumerable<TResult> GroupByAwait<TSource, TKey, TElement, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, UniTask<TResult>> resultSelector)
  108. {
  109. Error.ThrowArgumentNullException(source, nameof(source));
  110. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  111. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  112. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  113. return new GroupByAwait<TSource, TKey, TElement, TResult>(source, keySelector, elementSelector, resultSelector, EqualityComparer<TKey>.Default);
  114. }
  115. public static IUniTaskAsyncEnumerable<TResult> GroupByAwait<TSource, TKey, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TKey, IEnumerable<TSource>, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer)
  116. {
  117. Error.ThrowArgumentNullException(source, nameof(source));
  118. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  119. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  120. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  121. return new GroupByAwait<TSource, TKey, TSource, TResult>(source, keySelector, x => UniTask.FromResult(x), resultSelector, comparer);
  122. }
  123. public static IUniTaskAsyncEnumerable<TResult> GroupByAwait<TSource, TKey, TElement, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer)
  124. {
  125. Error.ThrowArgumentNullException(source, nameof(source));
  126. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  127. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  128. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  129. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  130. return new GroupByAwait<TSource, TKey, TElement, TResult>(source, keySelector, elementSelector, resultSelector, comparer);
  131. }
  132. // with ct
  133. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TSource>> GroupByAwaitWithCancellation<TSource, TKey>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector)
  134. {
  135. Error.ThrowArgumentNullException(source, nameof(source));
  136. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  137. return new GroupByAwaitWithCancellation<TSource, TKey, TSource>(source, keySelector, (x, _) => UniTask.FromResult(x), EqualityComparer<TKey>.Default);
  138. }
  139. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TSource>> GroupByAwaitWithCancellation<TSource, TKey>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, IEqualityComparer<TKey> comparer)
  140. {
  141. Error.ThrowArgumentNullException(source, nameof(source));
  142. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  143. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  144. return new GroupByAwaitWithCancellation<TSource, TKey, TSource>(source, keySelector, (x, _) => UniTask.FromResult(x), comparer);
  145. }
  146. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>> GroupByAwaitWithCancellation<TSource, TKey, TElement>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector)
  147. {
  148. Error.ThrowArgumentNullException(source, nameof(source));
  149. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  150. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  151. return new GroupByAwaitWithCancellation<TSource, TKey, TElement>(source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
  152. }
  153. public static IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>> GroupByAwaitWithCancellation<TSource, TKey, TElement>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, IEqualityComparer<TKey> comparer)
  154. {
  155. Error.ThrowArgumentNullException(source, nameof(source));
  156. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  157. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  158. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  159. return new GroupByAwaitWithCancellation<TSource, TKey, TElement>(source, keySelector, elementSelector, comparer);
  160. }
  161. public static IUniTaskAsyncEnumerable<TResult> GroupByAwaitWithCancellation<TSource, TKey, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TKey, IEnumerable<TSource>, CancellationToken, UniTask<TResult>> resultSelector)
  162. {
  163. Error.ThrowArgumentNullException(source, nameof(source));
  164. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  165. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  166. return new GroupByAwaitWithCancellation<TSource, TKey, TSource, TResult>(source, keySelector, (x, _) => UniTask.FromResult(x), resultSelector, EqualityComparer<TKey>.Default);
  167. }
  168. public static IUniTaskAsyncEnumerable<TResult> GroupByAwaitWithCancellation<TSource, TKey, TElement, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, CancellationToken, UniTask<TResult>> resultSelector)
  169. {
  170. Error.ThrowArgumentNullException(source, nameof(source));
  171. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  172. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  173. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  174. return new GroupByAwaitWithCancellation<TSource, TKey, TElement, TResult>(source, keySelector, elementSelector, resultSelector, EqualityComparer<TKey>.Default);
  175. }
  176. public static IUniTaskAsyncEnumerable<TResult> GroupByAwaitWithCancellation<TSource, TKey, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TKey, IEnumerable<TSource>, CancellationToken, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer)
  177. {
  178. Error.ThrowArgumentNullException(source, nameof(source));
  179. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  180. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  181. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  182. return new GroupByAwaitWithCancellation<TSource, TKey, TSource, TResult>(source, keySelector, (x, _) => UniTask.FromResult(x), resultSelector, comparer);
  183. }
  184. public static IUniTaskAsyncEnumerable<TResult> GroupByAwaitWithCancellation<TSource, TKey, TElement, TResult>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, CancellationToken, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer)
  185. {
  186. Error.ThrowArgumentNullException(source, nameof(source));
  187. Error.ThrowArgumentNullException(keySelector, nameof(keySelector));
  188. Error.ThrowArgumentNullException(elementSelector, nameof(elementSelector));
  189. Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
  190. Error.ThrowArgumentNullException(comparer, nameof(comparer));
  191. return new GroupByAwaitWithCancellation<TSource, TKey, TElement, TResult>(source, keySelector, elementSelector, resultSelector, comparer);
  192. }
  193. }
  194. internal sealed class GroupBy<TSource, TKey, TElement> : IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>>
  195. {
  196. readonly IUniTaskAsyncEnumerable<TSource> source;
  197. readonly Func<TSource, TKey> keySelector;
  198. readonly Func<TSource, TElement> elementSelector;
  199. readonly IEqualityComparer<TKey> comparer;
  200. public GroupBy(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, IEqualityComparer<TKey> comparer)
  201. {
  202. this.source = source;
  203. this.keySelector = keySelector;
  204. this.elementSelector = elementSelector;
  205. this.comparer = comparer;
  206. }
  207. public IUniTaskAsyncEnumerator<IGrouping<TKey, TElement>> GetAsyncEnumerator(CancellationToken cancellationToken = default)
  208. {
  209. return new _GroupBy(source, keySelector, elementSelector, comparer, cancellationToken);
  210. }
  211. sealed class _GroupBy : MoveNextSource, IUniTaskAsyncEnumerator<IGrouping<TKey, TElement>>
  212. {
  213. readonly IUniTaskAsyncEnumerable<TSource> source;
  214. readonly Func<TSource, TKey> keySelector;
  215. readonly Func<TSource, TElement> elementSelector;
  216. readonly IEqualityComparer<TKey> comparer;
  217. CancellationToken cancellationToken;
  218. IEnumerator<IGrouping<TKey, TElement>> groupEnumerator;
  219. public _GroupBy(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, IEqualityComparer<TKey> comparer, CancellationToken cancellationToken)
  220. {
  221. this.source = source;
  222. this.keySelector = keySelector;
  223. this.elementSelector = elementSelector;
  224. this.comparer = comparer;
  225. this.cancellationToken = cancellationToken;
  226. TaskTracker.TrackActiveTask(this, 3);
  227. }
  228. public IGrouping<TKey, TElement> Current { get; private set; }
  229. public UniTask<bool> MoveNextAsync()
  230. {
  231. cancellationToken.ThrowIfCancellationRequested();
  232. completionSource.Reset();
  233. if (groupEnumerator == null)
  234. {
  235. CreateLookup().Forget();
  236. }
  237. else
  238. {
  239. SourceMoveNext();
  240. }
  241. return new UniTask<bool>(this, completionSource.Version);
  242. }
  243. async UniTaskVoid CreateLookup()
  244. {
  245. try
  246. {
  247. var lookup = await source.ToLookupAsync(keySelector, elementSelector, comparer, cancellationToken);
  248. groupEnumerator = lookup.GetEnumerator();
  249. }
  250. catch (Exception ex)
  251. {
  252. completionSource.TrySetException(ex);
  253. return;
  254. }
  255. SourceMoveNext();
  256. }
  257. void SourceMoveNext()
  258. {
  259. try
  260. {
  261. if (groupEnumerator.MoveNext())
  262. {
  263. Current = groupEnumerator.Current as IGrouping<TKey, TElement>;
  264. completionSource.TrySetResult(true);
  265. }
  266. else
  267. {
  268. completionSource.TrySetResult(false);
  269. }
  270. }
  271. catch (Exception ex)
  272. {
  273. completionSource.TrySetException(ex);
  274. return;
  275. }
  276. }
  277. public UniTask DisposeAsync()
  278. {
  279. TaskTracker.RemoveTracking(this);
  280. if (groupEnumerator != null)
  281. {
  282. groupEnumerator.Dispose();
  283. }
  284. return default;
  285. }
  286. }
  287. }
  288. internal sealed class GroupBy<TSource, TKey, TElement, TResult> : IUniTaskAsyncEnumerable<TResult>
  289. {
  290. readonly IUniTaskAsyncEnumerable<TSource> source;
  291. readonly Func<TSource, TKey> keySelector;
  292. readonly Func<TSource, TElement> elementSelector;
  293. readonly Func<TKey, IEnumerable<TElement>, TResult> resultSelector;
  294. readonly IEqualityComparer<TKey> comparer;
  295. public GroupBy(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, Func<TKey, IEnumerable<TElement>, TResult> resultSelector, IEqualityComparer<TKey> comparer)
  296. {
  297. this.source = source;
  298. this.keySelector = keySelector;
  299. this.elementSelector = elementSelector;
  300. this.resultSelector = resultSelector;
  301. this.comparer = comparer;
  302. }
  303. public IUniTaskAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellationToken = default)
  304. {
  305. return new _GroupBy(source, keySelector, elementSelector, resultSelector, comparer, cancellationToken);
  306. }
  307. sealed class _GroupBy : MoveNextSource, IUniTaskAsyncEnumerator<TResult>
  308. {
  309. readonly IUniTaskAsyncEnumerable<TSource> source;
  310. readonly Func<TSource, TKey> keySelector;
  311. readonly Func<TSource, TElement> elementSelector;
  312. readonly Func<TKey, IEnumerable<TElement>, TResult> resultSelector;
  313. readonly IEqualityComparer<TKey> comparer;
  314. CancellationToken cancellationToken;
  315. IEnumerator<IGrouping<TKey, TElement>> groupEnumerator;
  316. public _GroupBy(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, Func<TKey, IEnumerable<TElement>, TResult> resultSelector, IEqualityComparer<TKey> comparer, CancellationToken cancellationToken)
  317. {
  318. this.source = source;
  319. this.keySelector = keySelector;
  320. this.elementSelector = elementSelector;
  321. this.resultSelector = resultSelector;
  322. this.comparer = comparer;
  323. this.cancellationToken = cancellationToken;
  324. TaskTracker.TrackActiveTask(this, 3);
  325. }
  326. public TResult Current { get; private set; }
  327. public UniTask<bool> MoveNextAsync()
  328. {
  329. cancellationToken.ThrowIfCancellationRequested();
  330. completionSource.Reset();
  331. if (groupEnumerator == null)
  332. {
  333. CreateLookup().Forget();
  334. }
  335. else
  336. {
  337. SourceMoveNext();
  338. }
  339. return new UniTask<bool>(this, completionSource.Version);
  340. }
  341. async UniTaskVoid CreateLookup()
  342. {
  343. try
  344. {
  345. var lookup = await source.ToLookupAsync(keySelector, elementSelector, comparer, cancellationToken);
  346. groupEnumerator = lookup.GetEnumerator();
  347. }
  348. catch (Exception ex)
  349. {
  350. completionSource.TrySetException(ex);
  351. return;
  352. }
  353. SourceMoveNext();
  354. }
  355. void SourceMoveNext()
  356. {
  357. try
  358. {
  359. if (groupEnumerator.MoveNext())
  360. {
  361. var current = groupEnumerator.Current;
  362. Current = resultSelector(current.Key, current);
  363. completionSource.TrySetResult(true);
  364. }
  365. else
  366. {
  367. completionSource.TrySetResult(false);
  368. }
  369. }
  370. catch (Exception ex)
  371. {
  372. completionSource.TrySetException(ex);
  373. return;
  374. }
  375. }
  376. public UniTask DisposeAsync()
  377. {
  378. TaskTracker.RemoveTracking(this);
  379. if (groupEnumerator != null)
  380. {
  381. groupEnumerator.Dispose();
  382. }
  383. return default;
  384. }
  385. }
  386. }
  387. internal sealed class GroupByAwait<TSource, TKey, TElement> : IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>>
  388. {
  389. readonly IUniTaskAsyncEnumerable<TSource> source;
  390. readonly Func<TSource, UniTask<TKey>> keySelector;
  391. readonly Func<TSource, UniTask<TElement>> elementSelector;
  392. readonly IEqualityComparer<TKey> comparer;
  393. public GroupByAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, IEqualityComparer<TKey> comparer)
  394. {
  395. this.source = source;
  396. this.keySelector = keySelector;
  397. this.elementSelector = elementSelector;
  398. this.comparer = comparer;
  399. }
  400. public IUniTaskAsyncEnumerator<IGrouping<TKey, TElement>> GetAsyncEnumerator(CancellationToken cancellationToken = default)
  401. {
  402. return new _GroupByAwait(source, keySelector, elementSelector, comparer, cancellationToken);
  403. }
  404. sealed class _GroupByAwait : MoveNextSource, IUniTaskAsyncEnumerator<IGrouping<TKey, TElement>>
  405. {
  406. readonly IUniTaskAsyncEnumerable<TSource> source;
  407. readonly Func<TSource, UniTask<TKey>> keySelector;
  408. readonly Func<TSource, UniTask<TElement>> elementSelector;
  409. readonly IEqualityComparer<TKey> comparer;
  410. CancellationToken cancellationToken;
  411. IEnumerator<IGrouping<TKey, TElement>> groupEnumerator;
  412. public _GroupByAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, IEqualityComparer<TKey> comparer, CancellationToken cancellationToken)
  413. {
  414. this.source = source;
  415. this.keySelector = keySelector;
  416. this.elementSelector = elementSelector;
  417. this.comparer = comparer;
  418. this.cancellationToken = cancellationToken;
  419. TaskTracker.TrackActiveTask(this, 3);
  420. }
  421. public IGrouping<TKey, TElement> Current { get; private set; }
  422. public UniTask<bool> MoveNextAsync()
  423. {
  424. cancellationToken.ThrowIfCancellationRequested();
  425. completionSource.Reset();
  426. if (groupEnumerator == null)
  427. {
  428. CreateLookup().Forget();
  429. }
  430. else
  431. {
  432. SourceMoveNext();
  433. }
  434. return new UniTask<bool>(this, completionSource.Version);
  435. }
  436. async UniTaskVoid CreateLookup()
  437. {
  438. try
  439. {
  440. var lookup = await source.ToLookupAwaitAsync(keySelector, elementSelector, comparer, cancellationToken);
  441. groupEnumerator = lookup.GetEnumerator();
  442. }
  443. catch (Exception ex)
  444. {
  445. completionSource.TrySetException(ex);
  446. return;
  447. }
  448. SourceMoveNext();
  449. }
  450. void SourceMoveNext()
  451. {
  452. try
  453. {
  454. if (groupEnumerator.MoveNext())
  455. {
  456. Current = groupEnumerator.Current as IGrouping<TKey, TElement>;
  457. completionSource.TrySetResult(true);
  458. }
  459. else
  460. {
  461. completionSource.TrySetResult(false);
  462. }
  463. }
  464. catch (Exception ex)
  465. {
  466. completionSource.TrySetException(ex);
  467. return;
  468. }
  469. }
  470. public UniTask DisposeAsync()
  471. {
  472. TaskTracker.RemoveTracking(this);
  473. if (groupEnumerator != null)
  474. {
  475. groupEnumerator.Dispose();
  476. }
  477. return default;
  478. }
  479. }
  480. }
  481. internal sealed class GroupByAwait<TSource, TKey, TElement, TResult> : IUniTaskAsyncEnumerable<TResult>
  482. {
  483. readonly IUniTaskAsyncEnumerable<TSource> source;
  484. readonly Func<TSource, UniTask<TKey>> keySelector;
  485. readonly Func<TSource, UniTask<TElement>> elementSelector;
  486. readonly Func<TKey, IEnumerable<TElement>, UniTask<TResult>> resultSelector;
  487. readonly IEqualityComparer<TKey> comparer;
  488. public GroupByAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer)
  489. {
  490. this.source = source;
  491. this.keySelector = keySelector;
  492. this.elementSelector = elementSelector;
  493. this.resultSelector = resultSelector;
  494. this.comparer = comparer;
  495. }
  496. public IUniTaskAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellationToken = default)
  497. {
  498. return new _GroupByAwait(source, keySelector, elementSelector, resultSelector, comparer, cancellationToken);
  499. }
  500. sealed class _GroupByAwait : MoveNextSource, IUniTaskAsyncEnumerator<TResult>
  501. {
  502. readonly static Action<object> ResultSelectCoreDelegate = ResultSelectCore;
  503. readonly IUniTaskAsyncEnumerable<TSource> source;
  504. readonly Func<TSource, UniTask<TKey>> keySelector;
  505. readonly Func<TSource, UniTask<TElement>> elementSelector;
  506. readonly Func<TKey, IEnumerable<TElement>, UniTask<TResult>> resultSelector;
  507. readonly IEqualityComparer<TKey> comparer;
  508. CancellationToken cancellationToken;
  509. IEnumerator<IGrouping<TKey, TElement>> groupEnumerator;
  510. UniTask<TResult>.Awaiter awaiter;
  511. public _GroupByAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<TKey>> keySelector, Func<TSource, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer, CancellationToken cancellationToken)
  512. {
  513. this.source = source;
  514. this.keySelector = keySelector;
  515. this.elementSelector = elementSelector;
  516. this.resultSelector = resultSelector;
  517. this.comparer = comparer;
  518. this.cancellationToken = cancellationToken;
  519. TaskTracker.TrackActiveTask(this, 3);
  520. }
  521. public TResult Current { get; private set; }
  522. public UniTask<bool> MoveNextAsync()
  523. {
  524. cancellationToken.ThrowIfCancellationRequested();
  525. completionSource.Reset();
  526. if (groupEnumerator == null)
  527. {
  528. CreateLookup().Forget();
  529. }
  530. else
  531. {
  532. SourceMoveNext();
  533. }
  534. return new UniTask<bool>(this, completionSource.Version);
  535. }
  536. async UniTaskVoid CreateLookup()
  537. {
  538. try
  539. {
  540. var lookup = await source.ToLookupAwaitAsync(keySelector, elementSelector, comparer, cancellationToken);
  541. groupEnumerator = lookup.GetEnumerator();
  542. }
  543. catch (Exception ex)
  544. {
  545. completionSource.TrySetException(ex);
  546. return;
  547. }
  548. SourceMoveNext();
  549. }
  550. void SourceMoveNext()
  551. {
  552. try
  553. {
  554. if (groupEnumerator.MoveNext())
  555. {
  556. var current = groupEnumerator.Current;
  557. awaiter = resultSelector(current.Key, current).GetAwaiter();
  558. if (awaiter.IsCompleted)
  559. {
  560. ResultSelectCore(this);
  561. }
  562. else
  563. {
  564. awaiter.SourceOnCompleted(ResultSelectCoreDelegate, this);
  565. }
  566. return;
  567. }
  568. else
  569. {
  570. completionSource.TrySetResult(false);
  571. }
  572. }
  573. catch (Exception ex)
  574. {
  575. completionSource.TrySetException(ex);
  576. return;
  577. }
  578. }
  579. static void ResultSelectCore(object state)
  580. {
  581. var self = (_GroupByAwait)state;
  582. if (self.TryGetResult(self.awaiter, out var result))
  583. {
  584. self.Current = result;
  585. self.completionSource.TrySetResult(true);
  586. }
  587. }
  588. public UniTask DisposeAsync()
  589. {
  590. TaskTracker.RemoveTracking(this);
  591. if (groupEnumerator != null)
  592. {
  593. groupEnumerator.Dispose();
  594. }
  595. return default;
  596. }
  597. }
  598. }
  599. internal sealed class GroupByAwaitWithCancellation<TSource, TKey, TElement> : IUniTaskAsyncEnumerable<IGrouping<TKey, TElement>>
  600. {
  601. readonly IUniTaskAsyncEnumerable<TSource> source;
  602. readonly Func<TSource, CancellationToken, UniTask<TKey>> keySelector;
  603. readonly Func<TSource, CancellationToken, UniTask<TElement>> elementSelector;
  604. readonly IEqualityComparer<TKey> comparer;
  605. public GroupByAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, IEqualityComparer<TKey> comparer)
  606. {
  607. this.source = source;
  608. this.keySelector = keySelector;
  609. this.elementSelector = elementSelector;
  610. this.comparer = comparer;
  611. }
  612. public IUniTaskAsyncEnumerator<IGrouping<TKey, TElement>> GetAsyncEnumerator(CancellationToken cancellationToken = default)
  613. {
  614. return new _GroupByAwaitWithCancellation(source, keySelector, elementSelector, comparer, cancellationToken);
  615. }
  616. sealed class _GroupByAwaitWithCancellation : MoveNextSource, IUniTaskAsyncEnumerator<IGrouping<TKey, TElement>>
  617. {
  618. readonly IUniTaskAsyncEnumerable<TSource> source;
  619. readonly Func<TSource, CancellationToken, UniTask<TKey>> keySelector;
  620. readonly Func<TSource, CancellationToken, UniTask<TElement>> elementSelector;
  621. readonly IEqualityComparer<TKey> comparer;
  622. CancellationToken cancellationToken;
  623. IEnumerator<IGrouping<TKey, TElement>> groupEnumerator;
  624. public _GroupByAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, IEqualityComparer<TKey> comparer, CancellationToken cancellationToken)
  625. {
  626. this.source = source;
  627. this.keySelector = keySelector;
  628. this.elementSelector = elementSelector;
  629. this.comparer = comparer;
  630. this.cancellationToken = cancellationToken;
  631. TaskTracker.TrackActiveTask(this, 3);
  632. }
  633. public IGrouping<TKey, TElement> Current { get; private set; }
  634. public UniTask<bool> MoveNextAsync()
  635. {
  636. cancellationToken.ThrowIfCancellationRequested();
  637. completionSource.Reset();
  638. if (groupEnumerator == null)
  639. {
  640. CreateLookup().Forget();
  641. }
  642. else
  643. {
  644. SourceMoveNext();
  645. }
  646. return new UniTask<bool>(this, completionSource.Version);
  647. }
  648. async UniTaskVoid CreateLookup()
  649. {
  650. try
  651. {
  652. var lookup = await source.ToLookupAwaitWithCancellationAsync(keySelector, elementSelector, comparer, cancellationToken);
  653. groupEnumerator = lookup.GetEnumerator();
  654. }
  655. catch (Exception ex)
  656. {
  657. completionSource.TrySetException(ex);
  658. return;
  659. }
  660. SourceMoveNext();
  661. }
  662. void SourceMoveNext()
  663. {
  664. try
  665. {
  666. if (groupEnumerator.MoveNext())
  667. {
  668. Current = groupEnumerator.Current as IGrouping<TKey, TElement>;
  669. completionSource.TrySetResult(true);
  670. }
  671. else
  672. {
  673. completionSource.TrySetResult(false);
  674. }
  675. }
  676. catch (Exception ex)
  677. {
  678. completionSource.TrySetException(ex);
  679. return;
  680. }
  681. }
  682. public UniTask DisposeAsync()
  683. {
  684. TaskTracker.RemoveTracking(this);
  685. if (groupEnumerator != null)
  686. {
  687. groupEnumerator.Dispose();
  688. }
  689. return default;
  690. }
  691. }
  692. }
  693. internal sealed class GroupByAwaitWithCancellation<TSource, TKey, TElement, TResult> : IUniTaskAsyncEnumerable<TResult>
  694. {
  695. readonly IUniTaskAsyncEnumerable<TSource> source;
  696. readonly Func<TSource, CancellationToken, UniTask<TKey>> keySelector;
  697. readonly Func<TSource, CancellationToken, UniTask<TElement>> elementSelector;
  698. readonly Func<TKey, IEnumerable<TElement>, CancellationToken, UniTask<TResult>> resultSelector;
  699. readonly IEqualityComparer<TKey> comparer;
  700. public GroupByAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, CancellationToken, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer)
  701. {
  702. this.source = source;
  703. this.keySelector = keySelector;
  704. this.elementSelector = elementSelector;
  705. this.resultSelector = resultSelector;
  706. this.comparer = comparer;
  707. }
  708. public IUniTaskAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellationToken = default)
  709. {
  710. return new _GroupByAwaitWithCancellation(source, keySelector, elementSelector, resultSelector, comparer, cancellationToken);
  711. }
  712. sealed class _GroupByAwaitWithCancellation : MoveNextSource, IUniTaskAsyncEnumerator<TResult>
  713. {
  714. readonly static Action<object> ResultSelectCoreDelegate = ResultSelectCore;
  715. readonly IUniTaskAsyncEnumerable<TSource> source;
  716. readonly Func<TSource, CancellationToken, UniTask<TKey>> keySelector;
  717. readonly Func<TSource, CancellationToken, UniTask<TElement>> elementSelector;
  718. readonly Func<TKey, IEnumerable<TElement>, CancellationToken, UniTask<TResult>> resultSelector;
  719. readonly IEqualityComparer<TKey> comparer;
  720. CancellationToken cancellationToken;
  721. IEnumerator<IGrouping<TKey, TElement>> groupEnumerator;
  722. UniTask<TResult>.Awaiter awaiter;
  723. public _GroupByAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<TKey>> keySelector, Func<TSource, CancellationToken, UniTask<TElement>> elementSelector, Func<TKey, IEnumerable<TElement>, CancellationToken, UniTask<TResult>> resultSelector, IEqualityComparer<TKey> comparer, CancellationToken cancellationToken)
  724. {
  725. this.source = source;
  726. this.keySelector = keySelector;
  727. this.elementSelector = elementSelector;
  728. this.resultSelector = resultSelector;
  729. this.comparer = comparer;
  730. this.cancellationToken = cancellationToken;
  731. TaskTracker.TrackActiveTask(this, 3);
  732. }
  733. public TResult Current { get; private set; }
  734. public UniTask<bool> MoveNextAsync()
  735. {
  736. cancellationToken.ThrowIfCancellationRequested();
  737. completionSource.Reset();
  738. if (groupEnumerator == null)
  739. {
  740. CreateLookup().Forget();
  741. }
  742. else
  743. {
  744. SourceMoveNext();
  745. }
  746. return new UniTask<bool>(this, completionSource.Version);
  747. }
  748. async UniTaskVoid CreateLookup()
  749. {
  750. try
  751. {
  752. var lookup = await source.ToLookupAwaitWithCancellationAsync(keySelector, elementSelector, comparer, cancellationToken);
  753. groupEnumerator = lookup.GetEnumerator();
  754. }
  755. catch (Exception ex)
  756. {
  757. completionSource.TrySetException(ex);
  758. return;
  759. }
  760. SourceMoveNext();
  761. }
  762. void SourceMoveNext()
  763. {
  764. try
  765. {
  766. if (groupEnumerator.MoveNext())
  767. {
  768. var current = groupEnumerator.Current;
  769. awaiter = resultSelector(current.Key, current, cancellationToken).GetAwaiter();
  770. if (awaiter.IsCompleted)
  771. {
  772. ResultSelectCore(this);
  773. }
  774. else
  775. {
  776. awaiter.SourceOnCompleted(ResultSelectCoreDelegate, this);
  777. }
  778. return;
  779. }
  780. else
  781. {
  782. completionSource.TrySetResult(false);
  783. }
  784. }
  785. catch (Exception ex)
  786. {
  787. completionSource.TrySetException(ex);
  788. return;
  789. }
  790. }
  791. static void ResultSelectCore(object state)
  792. {
  793. var self = (_GroupByAwaitWithCancellation)state;
  794. if (self.TryGetResult(self.awaiter, out var result))
  795. {
  796. self.Current = result;
  797. self.completionSource.TrySetResult(true);
  798. }
  799. }
  800. public UniTask DisposeAsync()
  801. {
  802. TaskTracker.RemoveTracking(this);
  803. if (groupEnumerator != null)
  804. {
  805. groupEnumerator.Dispose();
  806. }
  807. return default;
  808. }
  809. }
  810. }
  811. }