Buffer.cs 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755
  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.Reactive.Concurrency;
  6. using System.Reactive.Disposables;
  7. using System.Threading;
  8. using System.Threading.Tasks;
  9. namespace System.Reactive.Linq
  10. {
  11. public partial class AsyncObservable
  12. {
  13. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, int count)
  14. {
  15. if (source == null)
  16. throw new ArgumentNullException(nameof(source));
  17. if (count <= 0)
  18. throw new ArgumentNullException(nameof(count));
  19. return CreateAsyncObservable<IList<TSource>>.From(
  20. source,
  21. count,
  22. static (source, count, observer) => source.SubscribeSafeAsync(AsyncObserver.Buffer(observer, count)));
  23. }
  24. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, int count, int skip)
  25. {
  26. if (source == null)
  27. throw new ArgumentNullException(nameof(source));
  28. if (count <= 0)
  29. throw new ArgumentNullException(nameof(count));
  30. if (skip <= 0)
  31. throw new ArgumentNullException(nameof(skip));
  32. return CreateAsyncObservable<IList<TSource>>.From(
  33. source,
  34. (count, skip),
  35. static (source, state, observer) => source.SubscribeSafeAsync(AsyncObserver.Buffer(observer, state.count, state.skip)));
  36. }
  37. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, TimeSpan timeSpan)
  38. {
  39. if (source == null)
  40. throw new ArgumentNullException(nameof(source));
  41. if (timeSpan < TimeSpan.Zero)
  42. throw new ArgumentNullException(nameof(timeSpan));
  43. return CreateAsyncObservable<IList<TSource>>.From(
  44. source,
  45. timeSpan,
  46. static async (source, timeSpan, observer) =>
  47. {
  48. var (sink, timer) = await AsyncObserver.Buffer(observer, timeSpan).ConfigureAwait(false);
  49. var subscription = await source.SubscribeSafeAsync(sink).ConfigureAwait(false);
  50. return StableCompositeAsyncDisposable.Create(subscription, timer);
  51. });
  52. }
  53. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, TimeSpan timeSpan, IAsyncScheduler scheduler)
  54. {
  55. if (source == null)
  56. throw new ArgumentNullException(nameof(source));
  57. if (timeSpan < TimeSpan.Zero)
  58. throw new ArgumentNullException(nameof(timeSpan));
  59. if (scheduler == null)
  60. throw new ArgumentNullException(nameof(scheduler));
  61. return CreateAsyncObservable<IList<TSource>>.From(
  62. source,
  63. (timeSpan, scheduler),
  64. static async (source, state, observer) =>
  65. {
  66. var (sink, timer) = await AsyncObserver.Buffer(observer, state.timeSpan, state.scheduler).ConfigureAwait(false);
  67. var subscription = await source.SubscribeSafeAsync(sink).ConfigureAwait(false);
  68. return StableCompositeAsyncDisposable.Create(subscription, timer);
  69. });
  70. }
  71. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, TimeSpan timeSpan, TimeSpan timeShift)
  72. {
  73. if (source == null)
  74. throw new ArgumentNullException(nameof(source));
  75. if (timeSpan < TimeSpan.Zero)
  76. throw new ArgumentNullException(nameof(timeSpan));
  77. if (timeShift < TimeSpan.Zero)
  78. throw new ArgumentNullException(nameof(timeShift));
  79. return CreateAsyncObservable<IList<TSource>>.From(
  80. source,
  81. (timeSpan, timeShift),
  82. static async (source, state, observer) =>
  83. {
  84. var (sink, timer) = await AsyncObserver.Buffer(observer, state.timeSpan, state.timeShift).ConfigureAwait(false);
  85. var subscription = await source.SubscribeSafeAsync(sink).ConfigureAwait(false);
  86. return StableCompositeAsyncDisposable.Create(subscription, timer);
  87. });
  88. }
  89. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, TimeSpan timeSpan, TimeSpan timeShift, IAsyncScheduler scheduler)
  90. {
  91. if (source == null)
  92. throw new ArgumentNullException(nameof(source));
  93. if (timeSpan < TimeSpan.Zero)
  94. throw new ArgumentNullException(nameof(timeSpan));
  95. if (timeShift < TimeSpan.Zero)
  96. throw new ArgumentNullException(nameof(timeShift));
  97. if (scheduler == null)
  98. throw new ArgumentNullException(nameof(scheduler));
  99. return CreateAsyncObservable<IList<TSource>>.From(
  100. source,
  101. (timeSpan, timeShift, scheduler),
  102. static async (source, state, observer) =>
  103. {
  104. var (sink, timer) = await AsyncObserver.Buffer(observer, state.timeSpan, state.timeShift, state.scheduler).ConfigureAwait(false);
  105. var subscription = await source.SubscribeSafeAsync(sink).ConfigureAwait(false);
  106. return StableCompositeAsyncDisposable.Create(subscription, timer);
  107. });
  108. }
  109. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, TimeSpan timeSpan, int count)
  110. {
  111. if (source == null)
  112. throw new ArgumentNullException(nameof(source));
  113. if (timeSpan < TimeSpan.Zero)
  114. throw new ArgumentNullException(nameof(timeSpan));
  115. if (count <= 0)
  116. throw new ArgumentNullException(nameof(count));
  117. return CreateAsyncObservable<IList<TSource>>.From(
  118. source,
  119. (timeSpan, count),
  120. static async (source, state, observer) =>
  121. {
  122. var (sink, timer) = await AsyncObserver.Buffer(observer, state.timeSpan, state.count).ConfigureAwait(false);
  123. var subscription = await source.SubscribeSafeAsync(sink).ConfigureAwait(false);
  124. return StableCompositeAsyncDisposable.Create(subscription, timer);
  125. });
  126. }
  127. public static IAsyncObservable<IList<TSource>> Buffer<TSource>(this IAsyncObservable<TSource> source, TimeSpan timeSpan, int count, IAsyncScheduler scheduler)
  128. {
  129. if (source == null)
  130. throw new ArgumentNullException(nameof(source));
  131. if (timeSpan < TimeSpan.Zero)
  132. throw new ArgumentNullException(nameof(timeSpan));
  133. if (count <= 0)
  134. throw new ArgumentNullException(nameof(count));
  135. if (scheduler == null)
  136. throw new ArgumentNullException(nameof(scheduler));
  137. return CreateAsyncObservable<IList<TSource>>.From(
  138. source,
  139. (timeSpan, count, scheduler),
  140. static async (source, state, observer) =>
  141. {
  142. var (sink, timer) = await AsyncObserver.Buffer(observer, state.timeSpan, state.count, state.scheduler).ConfigureAwait(false);
  143. var subscription = await source.SubscribeSafeAsync(sink).ConfigureAwait(false);
  144. return StableCompositeAsyncDisposable.Create(subscription, timer);
  145. });
  146. }
  147. public static IAsyncObservable<IList<TSource>> Buffer<TSource, TBufferBoundary>(this IAsyncObservable<TSource> source, IAsyncObservable<TBufferBoundary> bufferBoundaries)
  148. {
  149. if (source == null)
  150. throw new ArgumentNullException(nameof(source));
  151. if (bufferBoundaries == null)
  152. throw new ArgumentNullException(nameof(bufferBoundaries));
  153. return CreateAsyncObservable<IList<TSource>>.From(
  154. source,
  155. bufferBoundaries,
  156. static async (source, bufferBoundaries, observer) =>
  157. {
  158. var (sourceObserver, boundariesObserver) = AsyncObserver.Buffer<TSource, TBufferBoundary>(observer);
  159. var sourceSubscription = await source.SubscribeSafeAsync(sourceObserver).ConfigureAwait(false);
  160. var boundariesSubscription = await bufferBoundaries.SubscribeSafeAsync(boundariesObserver).ConfigureAwait(false);
  161. return StableCompositeAsyncDisposable.Create(sourceSubscription, boundariesSubscription);
  162. });
  163. }
  164. // REVIEW: This overload is inherited from Rx but arguably a bit esoteric as it doesn't provide context to the closing selector.
  165. public static IAsyncObservable<IList<TSource>> Buffer<TSource, TBufferClosing>(this IAsyncObservable<TSource> source, Func<IAsyncObservable<TBufferClosing>> bufferClosingSelector)
  166. {
  167. if (source == null)
  168. throw new ArgumentNullException(nameof(source));
  169. if (bufferClosingSelector == null)
  170. throw new ArgumentNullException(nameof(bufferClosingSelector));
  171. return CreateAsyncObservable<IList<TSource>>.From(
  172. source,
  173. bufferClosingSelector,
  174. static async (source, bufferClosingSelector, observer) =>
  175. {
  176. var (sourceObserver, closingDisposable) = await AsyncObserver.Buffer<TSource, TBufferClosing>(observer, bufferClosingSelector).ConfigureAwait(false);
  177. var sourceSubscription = await source.SubscribeSafeAsync(sourceObserver).ConfigureAwait(false);
  178. return StableCompositeAsyncDisposable.Create(sourceSubscription, closingDisposable);
  179. });
  180. }
  181. }
  182. public partial class AsyncObserver
  183. {
  184. public static IAsyncObserver<TSource> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, int count) => Buffer(observer, count, count);
  185. public static IAsyncObserver<TSource> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, int count, int skip)
  186. {
  187. if (observer == null)
  188. throw new ArgumentNullException(nameof(observer));
  189. if (count <= 0)
  190. throw new ArgumentNullException(nameof(count));
  191. if (skip <= 0)
  192. throw new ArgumentNullException(nameof(skip));
  193. var queue = new Queue<IList<TSource>>();
  194. var n = 0;
  195. void CreateBuffer() => queue.Enqueue(new List<TSource>());
  196. CreateBuffer();
  197. return Create<TSource>(
  198. async x =>
  199. {
  200. foreach (var buffer in queue)
  201. {
  202. buffer.Add(x);
  203. }
  204. var c = n - count + 1;
  205. if (c >= 0 && c % skip == 0)
  206. {
  207. var buffer = queue.Dequeue();
  208. if (buffer.Count > 0)
  209. {
  210. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  211. }
  212. }
  213. n++;
  214. if (n % skip == 0)
  215. {
  216. CreateBuffer();
  217. }
  218. },
  219. ex =>
  220. {
  221. while (queue.Count > 0)
  222. {
  223. queue.Dequeue().Clear();
  224. }
  225. return observer.OnErrorAsync(ex);
  226. },
  227. async () =>
  228. {
  229. while (queue.Count > 0)
  230. {
  231. var buffer = queue.Dequeue();
  232. if (buffer.Count > 0)
  233. {
  234. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  235. }
  236. }
  237. await observer.OnCompletedAsync().ConfigureAwait(false);
  238. }
  239. );
  240. }
  241. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, TimeSpan timeSpan) => Buffer(observer, timeSpan, TaskPoolAsyncScheduler.Default);
  242. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, TimeSpan timeSpan, IAsyncScheduler scheduler)
  243. {
  244. if (observer == null)
  245. throw new ArgumentNullException(nameof(observer));
  246. if (timeSpan < TimeSpan.Zero)
  247. throw new ArgumentNullException(nameof(timeSpan));
  248. if (scheduler == null)
  249. throw new ArgumentNullException(nameof(scheduler));
  250. return CoreAsync();
  251. async Task<(IAsyncObserver<TSource>, IAsyncDisposable)> CoreAsync()
  252. {
  253. var gate = new AsyncGate();
  254. var buffer = new List<TSource>();
  255. var sink = Create<TSource>(
  256. async x =>
  257. {
  258. using (await gate.LockAsync().ConfigureAwait(false))
  259. {
  260. buffer.Add(x);
  261. }
  262. },
  263. async ex =>
  264. {
  265. using (await gate.LockAsync().ConfigureAwait(false))
  266. {
  267. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  268. }
  269. },
  270. async () =>
  271. {
  272. using (await gate.LockAsync().ConfigureAwait(false))
  273. {
  274. if (buffer.Count > 0)
  275. {
  276. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  277. }
  278. await observer.OnCompletedAsync().ConfigureAwait(false);
  279. }
  280. }
  281. );
  282. var timer = await scheduler.ScheduleAsync(async ct =>
  283. {
  284. while (!ct.IsCancellationRequested)
  285. {
  286. using (await gate.LockAsync().ConfigureAwait(false))
  287. {
  288. if (buffer.Count > 0)
  289. {
  290. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  291. buffer = new List<TSource>();
  292. }
  293. }
  294. await scheduler.Delay(timeSpan, ct).RendezVous(scheduler, ct);
  295. }
  296. }, timeSpan).ConfigureAwait(false);
  297. return (sink, timer);
  298. };
  299. }
  300. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, TimeSpan timeSpan, TimeSpan timeShift) => Buffer(observer, timeSpan, timeShift, TaskPoolAsyncScheduler.Default);
  301. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, TimeSpan timeSpan, TimeSpan timeShift, IAsyncScheduler scheduler)
  302. {
  303. if (observer == null)
  304. throw new ArgumentNullException(nameof(observer));
  305. if (timeSpan < TimeSpan.Zero)
  306. throw new ArgumentNullException(nameof(timeSpan));
  307. if (timeShift < TimeSpan.Zero)
  308. throw new ArgumentNullException(nameof(timeShift));
  309. if (scheduler == null)
  310. throw new ArgumentNullException(nameof(scheduler));
  311. return CoreAsync();
  312. async Task<(IAsyncObserver<TSource>, IAsyncDisposable)> CoreAsync()
  313. {
  314. var gate = new AsyncGate();
  315. var queue = new Queue<List<TSource>>();
  316. queue.Enqueue(new List<TSource>());
  317. var sink = Create<TSource>(
  318. async x =>
  319. {
  320. using (await gate.LockAsync().ConfigureAwait(false))
  321. {
  322. foreach (var buffer in queue)
  323. {
  324. buffer.Add(x);
  325. }
  326. }
  327. },
  328. async ex =>
  329. {
  330. using (await gate.LockAsync().ConfigureAwait(false))
  331. {
  332. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  333. }
  334. },
  335. async () =>
  336. {
  337. using (await gate.LockAsync().ConfigureAwait(false))
  338. {
  339. while (queue.Count > 0)
  340. {
  341. var buffer = queue.Dequeue();
  342. if (buffer.Count > 0)
  343. {
  344. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  345. }
  346. }
  347. await observer.OnCompletedAsync().ConfigureAwait(false);
  348. }
  349. }
  350. );
  351. var nextOpen = timeShift;
  352. var nextClose = timeSpan;
  353. var totalTime = TimeSpan.Zero;
  354. var isOpen = false;
  355. var isClose = false;
  356. TimeSpan GetNextDue()
  357. {
  358. if (nextOpen == nextClose)
  359. {
  360. isOpen = isClose = true;
  361. }
  362. else if (nextClose < nextOpen)
  363. {
  364. isClose = true;
  365. isOpen = false;
  366. }
  367. else
  368. {
  369. isOpen = true;
  370. isClose = false;
  371. }
  372. var newTotalTime = isClose ? nextClose : nextOpen;
  373. var due = newTotalTime - totalTime;
  374. totalTime = newTotalTime;
  375. if (isOpen)
  376. {
  377. nextOpen += timeShift;
  378. }
  379. if (isClose)
  380. {
  381. nextClose += timeShift;
  382. }
  383. return due;
  384. }
  385. var timer = await scheduler.ScheduleAsync(async ct =>
  386. {
  387. while (!ct.IsCancellationRequested)
  388. {
  389. using (await gate.LockAsync().ConfigureAwait(false))
  390. {
  391. if (isClose)
  392. {
  393. var buffer = queue.Dequeue();
  394. if (buffer.Count > 0)
  395. {
  396. await observer.OnNextAsync(buffer).RendezVous(scheduler, ct);
  397. }
  398. }
  399. if (isOpen)
  400. {
  401. queue.Enqueue(new List<TSource>());
  402. }
  403. }
  404. await scheduler.Delay(GetNextDue(), ct).RendezVous(scheduler, ct);
  405. }
  406. }, GetNextDue()).ConfigureAwait(false);
  407. return (sink, timer);
  408. };
  409. }
  410. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, TimeSpan timeSpan, int count) => Buffer(observer, timeSpan, count, TaskPoolAsyncScheduler.Default);
  411. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource>(IAsyncObserver<IList<TSource>> observer, TimeSpan timeSpan, int count, IAsyncScheduler scheduler)
  412. {
  413. if (observer == null)
  414. throw new ArgumentNullException(nameof(observer));
  415. if (timeSpan < TimeSpan.Zero)
  416. throw new ArgumentNullException(nameof(timeSpan));
  417. if (count <= 0)
  418. throw new ArgumentNullException(nameof(count));
  419. if (scheduler == null)
  420. throw new ArgumentNullException(nameof(scheduler));
  421. return CoreAsync();
  422. async Task<(IAsyncObserver<TSource>, IAsyncDisposable)> CoreAsync()
  423. {
  424. var gate = new AsyncGate();
  425. var timer = new SerialAsyncDisposable();
  426. var buffer = new List<TSource>();
  427. var n = 0;
  428. var id = 0;
  429. async Task CreateTimerAsync(int timerId)
  430. {
  431. var d = await scheduler.ScheduleAsync(async ct =>
  432. {
  433. using (await gate.LockAsync().ConfigureAwait(false))
  434. {
  435. if (timerId == id)
  436. {
  437. await FlushAsync().ConfigureAwait(false);
  438. }
  439. }
  440. }, timeSpan).ConfigureAwait(false);
  441. await timer.AssignAsync(d).ConfigureAwait(false);
  442. }
  443. async Task FlushAsync()
  444. {
  445. n = 0;
  446. ++id;
  447. var res = buffer;
  448. buffer = new List<TSource>();
  449. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  450. await CreateTimerAsync(id).ConfigureAwait(false);
  451. }
  452. var sink = Create<TSource>(
  453. async x =>
  454. {
  455. using (await gate.LockAsync().ConfigureAwait(false))
  456. {
  457. buffer.Add(x);
  458. if (++n == count)
  459. {
  460. await FlushAsync().ConfigureAwait(false);
  461. }
  462. }
  463. },
  464. async ex =>
  465. {
  466. using (await gate.LockAsync().ConfigureAwait(false))
  467. {
  468. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  469. }
  470. },
  471. async () =>
  472. {
  473. using (await gate.LockAsync().ConfigureAwait(false))
  474. {
  475. await observer.OnNextAsync(buffer).ConfigureAwait(false); // NB: We don't check for non-empty in sync Rx either.
  476. await observer.OnCompletedAsync().ConfigureAwait(false);
  477. }
  478. }
  479. );
  480. await CreateTimerAsync(0).ConfigureAwait(false);
  481. return (sink, timer);
  482. };
  483. }
  484. public static (IAsyncObserver<TSource>, IAsyncObserver<TBufferBoundary>) Buffer<TSource, TBufferBoundary>(IAsyncObserver<IList<TSource>> observer)
  485. {
  486. if (observer == null)
  487. throw new ArgumentNullException(nameof(observer));
  488. var gate = new AsyncGate();
  489. var buffer = new List<TSource>();
  490. return
  491. (
  492. Create<TSource>(
  493. async x =>
  494. {
  495. using (await gate.LockAsync().ConfigureAwait(false))
  496. {
  497. buffer.Add(x);
  498. }
  499. },
  500. async ex =>
  501. {
  502. using (await gate.LockAsync().ConfigureAwait(false))
  503. {
  504. buffer.Clear();
  505. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  506. }
  507. },
  508. async () =>
  509. {
  510. using (await gate.LockAsync().ConfigureAwait(false))
  511. {
  512. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  513. await observer.OnCompletedAsync().ConfigureAwait(false);
  514. }
  515. }
  516. ),
  517. Create<TBufferBoundary>(
  518. async x =>
  519. {
  520. using (await gate.LockAsync().ConfigureAwait(false))
  521. {
  522. var oldBuffer = buffer;
  523. buffer = new List<TSource>();
  524. await observer.OnNextAsync(oldBuffer).ConfigureAwait(false);
  525. }
  526. },
  527. async ex =>
  528. {
  529. using (await gate.LockAsync().ConfigureAwait(false))
  530. {
  531. buffer.Clear();
  532. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  533. }
  534. },
  535. async () =>
  536. {
  537. using (await gate.LockAsync().ConfigureAwait(false))
  538. {
  539. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  540. await observer.OnCompletedAsync().ConfigureAwait(false);
  541. }
  542. }
  543. )
  544. );
  545. }
  546. public static Task<(IAsyncObserver<TSource>, IAsyncDisposable)> Buffer<TSource, TBufferClosing>(IAsyncObserver<IList<TSource>> observer, Func<IAsyncObservable<TBufferClosing>> bufferClosingSelector)
  547. {
  548. if (observer == null)
  549. throw new ArgumentNullException(nameof(observer));
  550. if (bufferClosingSelector == null)
  551. throw new ArgumentNullException(nameof(bufferClosingSelector));
  552. return CoreAsync();
  553. async Task<(IAsyncObserver<TSource>, IAsyncDisposable)> CoreAsync()
  554. {
  555. var closeSubscription = new SerialAsyncDisposable();
  556. var gate = new AsyncGate();
  557. var queueLock = new AsyncQueueLock();
  558. var buffer = new List<TSource>();
  559. async ValueTask CreateBufferCloseAsync()
  560. {
  561. var closing = default(IAsyncObservable<TBufferClosing>);
  562. try
  563. {
  564. closing = bufferClosingSelector(); // REVIEW: Do we need an async variant?
  565. }
  566. catch (Exception ex)
  567. {
  568. using (await gate.LockAsync().ConfigureAwait(false))
  569. {
  570. buffer.Clear();
  571. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  572. }
  573. return;
  574. }
  575. var closingSubscription = new SingleAssignmentAsyncDisposable();
  576. await closeSubscription.AssignAsync(closingSubscription).ConfigureAwait(false);
  577. async ValueTask CloseBufferAsync()
  578. {
  579. await closingSubscription.DisposeAsync().ConfigureAwait(false);
  580. using (await gate.LockAsync().ConfigureAwait(false))
  581. {
  582. var oldBuffer = buffer;
  583. buffer = new List<TSource>();
  584. await observer.OnNextAsync(oldBuffer).ConfigureAwait(false);
  585. }
  586. await queueLock.WaitAsync(CreateBufferCloseAsync).ConfigureAwait(false);
  587. }
  588. var closingObserver =
  589. Create<TBufferClosing>(
  590. x => CloseBufferAsync(),
  591. async ex =>
  592. {
  593. using (await gate.LockAsync().ConfigureAwait(false))
  594. {
  595. buffer.Clear();
  596. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  597. }
  598. },
  599. CloseBufferAsync
  600. );
  601. var closingSubscriptionInner = await closing.SubscribeSafeAsync(closingObserver).ConfigureAwait(false);
  602. await closingSubscription.AssignAsync(closingSubscriptionInner).ConfigureAwait(false);
  603. }
  604. var sink =
  605. Create<TSource>(
  606. async x =>
  607. {
  608. using (await gate.LockAsync().ConfigureAwait(false))
  609. {
  610. buffer.Add(x);
  611. }
  612. },
  613. async ex =>
  614. {
  615. using (await gate.LockAsync().ConfigureAwait(false))
  616. {
  617. buffer.Clear();
  618. await observer.OnErrorAsync(ex).ConfigureAwait(false);
  619. }
  620. },
  621. async () =>
  622. {
  623. using (await gate.LockAsync().ConfigureAwait(false))
  624. {
  625. await observer.OnNextAsync(buffer).ConfigureAwait(false);
  626. await observer.OnCompletedAsync().ConfigureAwait(false);
  627. }
  628. }
  629. );
  630. await queueLock.WaitAsync(CreateBufferCloseAsync).ConfigureAwait(false);
  631. return (sink, closeSubscription);
  632. }
  633. }
  634. }
  635. }