Zip.cs 32 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004
  1. using System;
  2. using System.Collections.Generic;
  3. namespace UniRx.Operators
  4. {
  5. public delegate TR ZipFunc<T1, T2, T3, TR>(T1 arg1, T2 arg2, T3 arg3);
  6. public delegate TR ZipFunc<T1, T2, T3, T4, TR>(T1 arg1, T2 arg2, T3 arg3, T4 arg4);
  7. public delegate TR ZipFunc<T1, T2, T3, T4, T5, TR>(T1 arg1, T2 arg2, T3 arg3, T4 arg4, T5 arg5);
  8. public delegate TR ZipFunc<T1, T2, T3, T4, T5, T6, TR>(T1 arg1, T2 arg2, T3 arg3, T4 arg4, T5 arg5, T6 arg6);
  9. public delegate TR ZipFunc<T1, T2, T3, T4, T5, T6, T7, TR>(T1 arg1, T2 arg2, T3 arg3, T4 arg4, T5 arg5, T6 arg6, T7 arg7);
  10. // binary
  11. internal class ZipObservable<TLeft, TRight, TResult> : OperatorObservableBase<TResult>
  12. {
  13. readonly IObservable<TLeft> left;
  14. readonly IObservable<TRight> right;
  15. readonly Func<TLeft, TRight, TResult> selector;
  16. public ZipObservable(IObservable<TLeft> left, IObservable<TRight> right, Func<TLeft, TRight, TResult> selector)
  17. : base(left.IsRequiredSubscribeOnCurrentThread() || right.IsRequiredSubscribeOnCurrentThread())
  18. {
  19. this.left = left;
  20. this.right = right;
  21. this.selector = selector;
  22. }
  23. protected override IDisposable SubscribeCore(IObserver<TResult> observer, IDisposable cancel)
  24. {
  25. return new Zip(this, observer, cancel).Run();
  26. }
  27. class Zip : OperatorObserverBase<TResult, TResult>
  28. {
  29. readonly ZipObservable<TLeft, TRight, TResult> parent;
  30. readonly object gate = new object();
  31. readonly Queue<TLeft> leftQ = new Queue<TLeft>();
  32. bool leftCompleted = false;
  33. readonly Queue<TRight> rightQ = new Queue<TRight>();
  34. bool rightCompleted = false;
  35. public Zip(ZipObservable<TLeft, TRight, TResult> parent, IObserver<TResult> observer, IDisposable cancel)
  36. : base(observer, cancel)
  37. {
  38. this.parent = parent;
  39. }
  40. public IDisposable Run()
  41. {
  42. var l = parent.left.Subscribe(new LeftZipObserver(this));
  43. var r = parent.right.Subscribe(new RightZipObserver(this));
  44. return StableCompositeDisposable.Create(l, r, Disposable.Create(() =>
  45. {
  46. lock (gate)
  47. {
  48. leftQ.Clear();
  49. rightQ.Clear();
  50. }
  51. }));
  52. }
  53. // dequeue is in the lock
  54. void Dequeue()
  55. {
  56. TLeft lv;
  57. TRight rv;
  58. TResult v;
  59. if (leftQ.Count != 0 && rightQ.Count != 0)
  60. {
  61. lv = leftQ.Dequeue();
  62. rv = rightQ.Dequeue();
  63. }
  64. else if (leftCompleted || rightCompleted)
  65. {
  66. try { observer.OnCompleted(); }
  67. finally { Dispose(); }
  68. return;
  69. }
  70. else
  71. {
  72. return;
  73. }
  74. try
  75. {
  76. v = parent.selector(lv, rv);
  77. }
  78. catch (Exception ex)
  79. {
  80. try { observer.OnError(ex); }
  81. finally { Dispose(); }
  82. return;
  83. }
  84. OnNext(v);
  85. }
  86. public override void OnNext(TResult value)
  87. {
  88. base.observer.OnNext(value);
  89. }
  90. public override void OnError(Exception error)
  91. {
  92. try { observer.OnError(error); }
  93. finally { Dispose(); }
  94. }
  95. public override void OnCompleted()
  96. {
  97. try { observer.OnCompleted(); }
  98. finally { Dispose(); }
  99. }
  100. class LeftZipObserver : IObserver<TLeft>
  101. {
  102. readonly Zip parent;
  103. public LeftZipObserver(Zip parent)
  104. {
  105. this.parent = parent;
  106. }
  107. public void OnNext(TLeft value)
  108. {
  109. lock (parent.gate)
  110. {
  111. parent.leftQ.Enqueue(value);
  112. parent.Dequeue();
  113. }
  114. }
  115. public void OnError(Exception ex)
  116. {
  117. lock (parent.gate)
  118. {
  119. parent.OnError(ex);
  120. }
  121. }
  122. public void OnCompleted()
  123. {
  124. lock (parent.gate)
  125. {
  126. parent.leftCompleted = true;
  127. if (parent.rightCompleted) parent.OnCompleted();
  128. }
  129. }
  130. }
  131. class RightZipObserver : IObserver<TRight>
  132. {
  133. readonly Zip parent;
  134. public RightZipObserver(Zip parent)
  135. {
  136. this.parent = parent;
  137. }
  138. public void OnNext(TRight value)
  139. {
  140. lock (parent.gate)
  141. {
  142. parent.rightQ.Enqueue(value);
  143. parent.Dequeue();
  144. }
  145. }
  146. public void OnError(Exception ex)
  147. {
  148. lock (parent.gate)
  149. {
  150. parent.OnError(ex);
  151. }
  152. }
  153. public void OnCompleted()
  154. {
  155. lock (parent.gate)
  156. {
  157. parent.rightCompleted = true;
  158. if (parent.leftCompleted) parent.OnCompleted();
  159. }
  160. }
  161. }
  162. }
  163. }
  164. // array
  165. internal class ZipObservable<T> : OperatorObservableBase<IList<T>>
  166. {
  167. readonly IObservable<T>[] sources;
  168. public ZipObservable(IObservable<T>[] sources)
  169. : base(true)
  170. {
  171. this.sources = sources;
  172. }
  173. protected override IDisposable SubscribeCore(IObserver<IList<T>> observer, IDisposable cancel)
  174. {
  175. return new Zip(this, observer, cancel).Run();
  176. }
  177. class Zip : OperatorObserverBase<IList<T>, IList<T>>
  178. {
  179. readonly ZipObservable<T> parent;
  180. readonly object gate = new object();
  181. Queue<T>[] queues;
  182. bool[] isDone;
  183. int length;
  184. public Zip(ZipObservable<T> parent, IObserver<IList<T>> observer, IDisposable cancel) : base(observer, cancel)
  185. {
  186. this.parent = parent;
  187. }
  188. public IDisposable Run()
  189. {
  190. length = parent.sources.Length;
  191. queues = new Queue<T>[length];
  192. isDone = new bool[length];
  193. for (int i = 0; i < length; i++)
  194. {
  195. queues[i] = new Queue<T>();
  196. }
  197. var disposables = new IDisposable[length + 1];
  198. for (int i = 0; i < length; i++)
  199. {
  200. var source = parent.sources[i];
  201. disposables[i] = source.Subscribe(new ZipObserver(this, i));
  202. }
  203. disposables[length] = Disposable.Create(() =>
  204. {
  205. lock (gate)
  206. {
  207. for (int i = 0; i < length; i++)
  208. {
  209. var q = queues[i];
  210. q.Clear();
  211. }
  212. }
  213. });
  214. return StableCompositeDisposable.CreateUnsafe(disposables);
  215. }
  216. // dequeue is in the lock
  217. void Dequeue(int index)
  218. {
  219. var allQueueHasValue = true;
  220. for (int i = 0; i < length; i++)
  221. {
  222. if (queues[i].Count == 0)
  223. {
  224. allQueueHasValue = false;
  225. break;
  226. }
  227. }
  228. if (!allQueueHasValue)
  229. {
  230. var allCompletedWithoutSelf = true;
  231. for (int i = 0; i < length; i++)
  232. {
  233. if (i == index) continue;
  234. if (!isDone[i])
  235. {
  236. allCompletedWithoutSelf = false;
  237. break;
  238. }
  239. }
  240. if (allCompletedWithoutSelf)
  241. {
  242. try { observer.OnCompleted(); }
  243. finally { Dispose(); }
  244. return;
  245. }
  246. else
  247. {
  248. return;
  249. }
  250. }
  251. var array = new T[length];
  252. for (int i = 0; i < length; i++)
  253. {
  254. array[i] = queues[i].Dequeue();
  255. }
  256. OnNext(array);
  257. }
  258. public override void OnNext(IList<T> value)
  259. {
  260. base.observer.OnNext(value);
  261. }
  262. public override void OnError(Exception error)
  263. {
  264. try { observer.OnError(error); }
  265. finally { Dispose(); }
  266. }
  267. public override void OnCompleted()
  268. {
  269. try { observer.OnCompleted(); }
  270. finally { Dispose(); }
  271. }
  272. class ZipObserver : IObserver<T>
  273. {
  274. readonly Zip parent;
  275. readonly int index;
  276. public ZipObserver(Zip parent, int index)
  277. {
  278. this.parent = parent;
  279. this.index = index;
  280. }
  281. public void OnNext(T value)
  282. {
  283. lock (parent.gate)
  284. {
  285. parent.queues[index].Enqueue(value);
  286. parent.Dequeue(index);
  287. }
  288. }
  289. public void OnError(Exception ex)
  290. {
  291. lock (parent.gate)
  292. {
  293. parent.OnError(ex);
  294. }
  295. }
  296. public void OnCompleted()
  297. {
  298. lock (parent.gate)
  299. {
  300. parent.isDone[index] = true;
  301. var allTrue = true;
  302. for (int i = 0; i < parent.length; i++)
  303. {
  304. if (!parent.isDone[i])
  305. {
  306. allTrue = false;
  307. break;
  308. }
  309. }
  310. if (allTrue)
  311. {
  312. parent.OnCompleted();
  313. }
  314. }
  315. }
  316. }
  317. }
  318. }
  319. // Generated from UniRx.Console.ZipGenerator.tt
  320. #region NTH
  321. internal class ZipObservable<T1, T2, T3, TR> : OperatorObservableBase<TR>
  322. {
  323. IObservable<T1> source1;
  324. IObservable<T2> source2;
  325. IObservable<T3> source3;
  326. ZipFunc<T1, T2, T3, TR> resultSelector;
  327. public ZipObservable(
  328. IObservable<T1> source1,
  329. IObservable<T2> source2,
  330. IObservable<T3> source3,
  331. ZipFunc<T1, T2, T3, TR> resultSelector)
  332. : base(
  333. source1.IsRequiredSubscribeOnCurrentThread() ||
  334. source2.IsRequiredSubscribeOnCurrentThread() ||
  335. source3.IsRequiredSubscribeOnCurrentThread() ||
  336. false)
  337. {
  338. this.source1 = source1;
  339. this.source2 = source2;
  340. this.source3 = source3;
  341. this.resultSelector = resultSelector;
  342. }
  343. protected override IDisposable SubscribeCore(IObserver<TR> observer, IDisposable cancel)
  344. {
  345. return new Zip(this, observer, cancel).Run();
  346. }
  347. class Zip : NthZipObserverBase<TR>
  348. {
  349. readonly ZipObservable<T1, T2, T3, TR> parent;
  350. readonly object gate = new object();
  351. readonly Queue<T1> q1 = new Queue<T1>();
  352. readonly Queue<T2> q2 = new Queue<T2>();
  353. readonly Queue<T3> q3 = new Queue<T3>();
  354. public Zip(ZipObservable<T1, T2, T3, TR> parent, IObserver<TR> observer, IDisposable cancel)
  355. : base(observer, cancel)
  356. {
  357. this.parent = parent;
  358. }
  359. public IDisposable Run()
  360. {
  361. base.SetQueue(new System.Collections.ICollection[] { q1, q2, q3 });
  362. var s1 = parent.source1.Subscribe(new ZipObserver<T1>(gate, this, 0, q1));
  363. var s2 = parent.source2.Subscribe(new ZipObserver<T2>(gate, this, 1, q2));
  364. var s3 = parent.source3.Subscribe(new ZipObserver<T3>(gate, this, 2, q3));
  365. return StableCompositeDisposable.Create(s1, s2, s3, Disposable.Create(() =>
  366. {
  367. lock (gate)
  368. {
  369. q1.Clear(); q2.Clear(); q3.Clear();
  370. }
  371. }));
  372. }
  373. public override TR GetResult()
  374. {
  375. return parent.resultSelector(q1.Dequeue(), q2.Dequeue(), q3.Dequeue());
  376. }
  377. public override void OnNext(TR value)
  378. {
  379. base.observer.OnNext(value);
  380. }
  381. public override void OnError(Exception error)
  382. {
  383. try { observer.OnError(error); }
  384. finally { Dispose(); }
  385. }
  386. public override void OnCompleted()
  387. {
  388. try { observer.OnCompleted(); }
  389. finally { Dispose(); }
  390. }
  391. }
  392. }
  393. internal class ZipObservable<T1, T2, T3, T4, TR> : OperatorObservableBase<TR>
  394. {
  395. IObservable<T1> source1;
  396. IObservable<T2> source2;
  397. IObservable<T3> source3;
  398. IObservable<T4> source4;
  399. ZipFunc<T1, T2, T3, T4, TR> resultSelector;
  400. public ZipObservable(
  401. IObservable<T1> source1,
  402. IObservable<T2> source2,
  403. IObservable<T3> source3,
  404. IObservable<T4> source4,
  405. ZipFunc<T1, T2, T3, T4, TR> resultSelector)
  406. : base(
  407. source1.IsRequiredSubscribeOnCurrentThread() ||
  408. source2.IsRequiredSubscribeOnCurrentThread() ||
  409. source3.IsRequiredSubscribeOnCurrentThread() ||
  410. source4.IsRequiredSubscribeOnCurrentThread() ||
  411. false)
  412. {
  413. this.source1 = source1;
  414. this.source2 = source2;
  415. this.source3 = source3;
  416. this.source4 = source4;
  417. this.resultSelector = resultSelector;
  418. }
  419. protected override IDisposable SubscribeCore(IObserver<TR> observer, IDisposable cancel)
  420. {
  421. return new Zip(this, observer, cancel).Run();
  422. }
  423. class Zip : NthZipObserverBase<TR>
  424. {
  425. readonly ZipObservable<T1, T2, T3, T4, TR> parent;
  426. readonly object gate = new object();
  427. readonly Queue<T1> q1 = new Queue<T1>();
  428. readonly Queue<T2> q2 = new Queue<T2>();
  429. readonly Queue<T3> q3 = new Queue<T3>();
  430. readonly Queue<T4> q4 = new Queue<T4>();
  431. public Zip(ZipObservable<T1, T2, T3, T4, TR> parent, IObserver<TR> observer, IDisposable cancel)
  432. : base(observer, cancel)
  433. {
  434. this.parent = parent;
  435. }
  436. public IDisposable Run()
  437. {
  438. base.SetQueue(new System.Collections.ICollection[] { q1, q2, q3, q4 });
  439. var s1 = parent.source1.Subscribe(new ZipObserver<T1>(gate, this, 0, q1));
  440. var s2 = parent.source2.Subscribe(new ZipObserver<T2>(gate, this, 1, q2));
  441. var s3 = parent.source3.Subscribe(new ZipObserver<T3>(gate, this, 2, q3));
  442. var s4 = parent.source4.Subscribe(new ZipObserver<T4>(gate, this, 3, q4));
  443. return StableCompositeDisposable.Create(s1, s2, s3, s4, Disposable.Create(() =>
  444. {
  445. lock (gate)
  446. {
  447. q1.Clear(); q2.Clear(); q3.Clear(); q4.Clear();
  448. }
  449. }));
  450. }
  451. public override TR GetResult()
  452. {
  453. return parent.resultSelector(q1.Dequeue(), q2.Dequeue(), q3.Dequeue(), q4.Dequeue());
  454. }
  455. public override void OnNext(TR value)
  456. {
  457. base.observer.OnNext(value);
  458. }
  459. public override void OnError(Exception error)
  460. {
  461. try { observer.OnError(error); }
  462. finally { Dispose(); }
  463. }
  464. public override void OnCompleted()
  465. {
  466. try { observer.OnCompleted(); }
  467. finally { Dispose(); }
  468. }
  469. }
  470. }
  471. internal class ZipObservable<T1, T2, T3, T4, T5, TR> : OperatorObservableBase<TR>
  472. {
  473. IObservable<T1> source1;
  474. IObservable<T2> source2;
  475. IObservable<T3> source3;
  476. IObservable<T4> source4;
  477. IObservable<T5> source5;
  478. ZipFunc<T1, T2, T3, T4, T5, TR> resultSelector;
  479. public ZipObservable(
  480. IObservable<T1> source1,
  481. IObservable<T2> source2,
  482. IObservable<T3> source3,
  483. IObservable<T4> source4,
  484. IObservable<T5> source5,
  485. ZipFunc<T1, T2, T3, T4, T5, TR> resultSelector)
  486. : base(
  487. source1.IsRequiredSubscribeOnCurrentThread() ||
  488. source2.IsRequiredSubscribeOnCurrentThread() ||
  489. source3.IsRequiredSubscribeOnCurrentThread() ||
  490. source4.IsRequiredSubscribeOnCurrentThread() ||
  491. source5.IsRequiredSubscribeOnCurrentThread() ||
  492. false)
  493. {
  494. this.source1 = source1;
  495. this.source2 = source2;
  496. this.source3 = source3;
  497. this.source4 = source4;
  498. this.source5 = source5;
  499. this.resultSelector = resultSelector;
  500. }
  501. protected override IDisposable SubscribeCore(IObserver<TR> observer, IDisposable cancel)
  502. {
  503. return new Zip(this, observer, cancel).Run();
  504. }
  505. class Zip : NthZipObserverBase<TR>
  506. {
  507. readonly ZipObservable<T1, T2, T3, T4, T5, TR> parent;
  508. readonly object gate = new object();
  509. readonly Queue<T1> q1 = new Queue<T1>();
  510. readonly Queue<T2> q2 = new Queue<T2>();
  511. readonly Queue<T3> q3 = new Queue<T3>();
  512. readonly Queue<T4> q4 = new Queue<T4>();
  513. readonly Queue<T5> q5 = new Queue<T5>();
  514. public Zip(ZipObservable<T1, T2, T3, T4, T5, TR> parent, IObserver<TR> observer, IDisposable cancel)
  515. : base(observer, cancel)
  516. {
  517. this.parent = parent;
  518. }
  519. public IDisposable Run()
  520. {
  521. base.SetQueue(new System.Collections.ICollection[] { q1, q2, q3, q4, q5 });
  522. var s1 = parent.source1.Subscribe(new ZipObserver<T1>(gate, this, 0, q1));
  523. var s2 = parent.source2.Subscribe(new ZipObserver<T2>(gate, this, 1, q2));
  524. var s3 = parent.source3.Subscribe(new ZipObserver<T3>(gate, this, 2, q3));
  525. var s4 = parent.source4.Subscribe(new ZipObserver<T4>(gate, this, 3, q4));
  526. var s5 = parent.source5.Subscribe(new ZipObserver<T5>(gate, this, 4, q5));
  527. return StableCompositeDisposable.Create(s1, s2, s3, s4, s5, Disposable.Create(() =>
  528. {
  529. lock (gate)
  530. {
  531. q1.Clear(); q2.Clear(); q3.Clear(); q4.Clear(); q5.Clear();
  532. }
  533. }));
  534. }
  535. public override TR GetResult()
  536. {
  537. return parent.resultSelector(q1.Dequeue(), q2.Dequeue(), q3.Dequeue(), q4.Dequeue(), q5.Dequeue());
  538. }
  539. public override void OnNext(TR value)
  540. {
  541. base.observer.OnNext(value);
  542. }
  543. public override void OnError(Exception error)
  544. {
  545. try { observer.OnError(error); }
  546. finally { Dispose(); }
  547. }
  548. public override void OnCompleted()
  549. {
  550. try { observer.OnCompleted(); }
  551. finally { Dispose(); }
  552. }
  553. }
  554. }
  555. internal class ZipObservable<T1, T2, T3, T4, T5, T6, TR> : OperatorObservableBase<TR>
  556. {
  557. IObservable<T1> source1;
  558. IObservable<T2> source2;
  559. IObservable<T3> source3;
  560. IObservable<T4> source4;
  561. IObservable<T5> source5;
  562. IObservable<T6> source6;
  563. ZipFunc<T1, T2, T3, T4, T5, T6, TR> resultSelector;
  564. public ZipObservable(
  565. IObservable<T1> source1,
  566. IObservable<T2> source2,
  567. IObservable<T3> source3,
  568. IObservable<T4> source4,
  569. IObservable<T5> source5,
  570. IObservable<T6> source6,
  571. ZipFunc<T1, T2, T3, T4, T5, T6, TR> resultSelector)
  572. : base(
  573. source1.IsRequiredSubscribeOnCurrentThread() ||
  574. source2.IsRequiredSubscribeOnCurrentThread() ||
  575. source3.IsRequiredSubscribeOnCurrentThread() ||
  576. source4.IsRequiredSubscribeOnCurrentThread() ||
  577. source5.IsRequiredSubscribeOnCurrentThread() ||
  578. source6.IsRequiredSubscribeOnCurrentThread() ||
  579. false)
  580. {
  581. this.source1 = source1;
  582. this.source2 = source2;
  583. this.source3 = source3;
  584. this.source4 = source4;
  585. this.source5 = source5;
  586. this.source6 = source6;
  587. this.resultSelector = resultSelector;
  588. }
  589. protected override IDisposable SubscribeCore(IObserver<TR> observer, IDisposable cancel)
  590. {
  591. return new Zip(this, observer, cancel).Run();
  592. }
  593. class Zip : NthZipObserverBase<TR>
  594. {
  595. readonly ZipObservable<T1, T2, T3, T4, T5, T6, TR> parent;
  596. readonly object gate = new object();
  597. readonly Queue<T1> q1 = new Queue<T1>();
  598. readonly Queue<T2> q2 = new Queue<T2>();
  599. readonly Queue<T3> q3 = new Queue<T3>();
  600. readonly Queue<T4> q4 = new Queue<T4>();
  601. readonly Queue<T5> q5 = new Queue<T5>();
  602. readonly Queue<T6> q6 = new Queue<T6>();
  603. public Zip(ZipObservable<T1, T2, T3, T4, T5, T6, TR> parent, IObserver<TR> observer, IDisposable cancel)
  604. : base(observer, cancel)
  605. {
  606. this.parent = parent;
  607. }
  608. public IDisposable Run()
  609. {
  610. base.SetQueue(new System.Collections.ICollection[] { q1, q2, q3, q4, q5, q6 });
  611. var s1 = parent.source1.Subscribe(new ZipObserver<T1>(gate, this, 0, q1));
  612. var s2 = parent.source2.Subscribe(new ZipObserver<T2>(gate, this, 1, q2));
  613. var s3 = parent.source3.Subscribe(new ZipObserver<T3>(gate, this, 2, q3));
  614. var s4 = parent.source4.Subscribe(new ZipObserver<T4>(gate, this, 3, q4));
  615. var s5 = parent.source5.Subscribe(new ZipObserver<T5>(gate, this, 4, q5));
  616. var s6 = parent.source6.Subscribe(new ZipObserver<T6>(gate, this, 5, q6));
  617. return StableCompositeDisposable.Create(s1, s2, s3, s4, s5, s6, Disposable.Create(() =>
  618. {
  619. lock (gate)
  620. {
  621. q1.Clear(); q2.Clear(); q3.Clear(); q4.Clear(); q5.Clear(); q6.Clear();
  622. }
  623. }));
  624. }
  625. public override TR GetResult()
  626. {
  627. return parent.resultSelector(q1.Dequeue(), q2.Dequeue(), q3.Dequeue(), q4.Dequeue(), q5.Dequeue(), q6.Dequeue());
  628. }
  629. public override void OnNext(TR value)
  630. {
  631. base.observer.OnNext(value);
  632. }
  633. public override void OnError(Exception error)
  634. {
  635. try { observer.OnError(error); }
  636. finally { Dispose(); }
  637. }
  638. public override void OnCompleted()
  639. {
  640. try { observer.OnCompleted(); }
  641. finally { Dispose(); }
  642. }
  643. }
  644. }
  645. internal class ZipObservable<T1, T2, T3, T4, T5, T6, T7, TR> : OperatorObservableBase<TR>
  646. {
  647. IObservable<T1> source1;
  648. IObservable<T2> source2;
  649. IObservable<T3> source3;
  650. IObservable<T4> source4;
  651. IObservable<T5> source5;
  652. IObservable<T6> source6;
  653. IObservable<T7> source7;
  654. ZipFunc<T1, T2, T3, T4, T5, T6, T7, TR> resultSelector;
  655. public ZipObservable(
  656. IObservable<T1> source1,
  657. IObservable<T2> source2,
  658. IObservable<T3> source3,
  659. IObservable<T4> source4,
  660. IObservable<T5> source5,
  661. IObservable<T6> source6,
  662. IObservable<T7> source7,
  663. ZipFunc<T1, T2, T3, T4, T5, T6, T7, TR> resultSelector)
  664. : base(
  665. source1.IsRequiredSubscribeOnCurrentThread() ||
  666. source2.IsRequiredSubscribeOnCurrentThread() ||
  667. source3.IsRequiredSubscribeOnCurrentThread() ||
  668. source4.IsRequiredSubscribeOnCurrentThread() ||
  669. source5.IsRequiredSubscribeOnCurrentThread() ||
  670. source6.IsRequiredSubscribeOnCurrentThread() ||
  671. source7.IsRequiredSubscribeOnCurrentThread() ||
  672. false)
  673. {
  674. this.source1 = source1;
  675. this.source2 = source2;
  676. this.source3 = source3;
  677. this.source4 = source4;
  678. this.source5 = source5;
  679. this.source6 = source6;
  680. this.source7 = source7;
  681. this.resultSelector = resultSelector;
  682. }
  683. protected override IDisposable SubscribeCore(IObserver<TR> observer, IDisposable cancel)
  684. {
  685. return new Zip(this, observer, cancel).Run();
  686. }
  687. class Zip : NthZipObserverBase<TR>
  688. {
  689. readonly ZipObservable<T1, T2, T3, T4, T5, T6, T7, TR> parent;
  690. readonly object gate = new object();
  691. readonly Queue<T1> q1 = new Queue<T1>();
  692. readonly Queue<T2> q2 = new Queue<T2>();
  693. readonly Queue<T3> q3 = new Queue<T3>();
  694. readonly Queue<T4> q4 = new Queue<T4>();
  695. readonly Queue<T5> q5 = new Queue<T5>();
  696. readonly Queue<T6> q6 = new Queue<T6>();
  697. readonly Queue<T7> q7 = new Queue<T7>();
  698. public Zip(ZipObservable<T1, T2, T3, T4, T5, T6, T7, TR> parent, IObserver<TR> observer, IDisposable cancel)
  699. : base(observer, cancel)
  700. {
  701. this.parent = parent;
  702. }
  703. public IDisposable Run()
  704. {
  705. base.SetQueue(new System.Collections.ICollection[] { q1, q2, q3, q4, q5, q6, q7 });
  706. var s1 = parent.source1.Subscribe(new ZipObserver<T1>(gate, this, 0, q1));
  707. var s2 = parent.source2.Subscribe(new ZipObserver<T2>(gate, this, 1, q2));
  708. var s3 = parent.source3.Subscribe(new ZipObserver<T3>(gate, this, 2, q3));
  709. var s4 = parent.source4.Subscribe(new ZipObserver<T4>(gate, this, 3, q4));
  710. var s5 = parent.source5.Subscribe(new ZipObserver<T5>(gate, this, 4, q5));
  711. var s6 = parent.source6.Subscribe(new ZipObserver<T6>(gate, this, 5, q6));
  712. var s7 = parent.source7.Subscribe(new ZipObserver<T7>(gate, this, 6, q7));
  713. return StableCompositeDisposable.Create(s1, s2, s3, s4, s5, s6, s7, Disposable.Create(() =>
  714. {
  715. lock (gate)
  716. {
  717. q1.Clear(); q2.Clear(); q3.Clear(); q4.Clear(); q5.Clear(); q6.Clear(); q7.Clear();
  718. }
  719. }));
  720. }
  721. public override TR GetResult()
  722. {
  723. return parent.resultSelector(q1.Dequeue(), q2.Dequeue(), q3.Dequeue(), q4.Dequeue(), q5.Dequeue(), q6.Dequeue(), q7.Dequeue());
  724. }
  725. public override void OnNext(TR value)
  726. {
  727. base.observer.OnNext(value);
  728. }
  729. public override void OnError(Exception error)
  730. {
  731. try { observer.OnError(error); }
  732. finally { Dispose(); }
  733. }
  734. public override void OnCompleted()
  735. {
  736. try { observer.OnCompleted(); }
  737. finally { Dispose(); }
  738. }
  739. }
  740. }
  741. #endregion
  742. // Nth infrastructure
  743. internal interface IZipObservable
  744. {
  745. void Dequeue(int index);
  746. void Fail(Exception error);
  747. void Done(int index);
  748. }
  749. internal abstract class NthZipObserverBase<T> : OperatorObserverBase<T, T>, IZipObservable
  750. {
  751. System.Collections.ICollection[] queues;
  752. bool[] isDone;
  753. int length;
  754. public NthZipObserverBase(IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  755. {
  756. }
  757. protected void SetQueue(System.Collections.ICollection[] queues)
  758. {
  759. this.queues = queues;
  760. this.length = queues.Length;
  761. this.isDone = new bool[length];
  762. }
  763. public abstract T GetResult();
  764. // operators in lock
  765. public void Dequeue(int index)
  766. {
  767. var allQueueHasValue = true;
  768. for (int i = 0; i < length; i++)
  769. {
  770. if (queues[i].Count == 0)
  771. {
  772. allQueueHasValue = false;
  773. break;
  774. }
  775. }
  776. if (!allQueueHasValue)
  777. {
  778. var allCompletedWithoutSelf = true;
  779. for (int i = 0; i < length; i++)
  780. {
  781. if (i == index) continue;
  782. if (!isDone[i])
  783. {
  784. allCompletedWithoutSelf = false;
  785. break;
  786. }
  787. }
  788. if (allCompletedWithoutSelf)
  789. {
  790. try { observer.OnCompleted(); }
  791. finally { Dispose(); }
  792. return;
  793. }
  794. else
  795. {
  796. return;
  797. }
  798. }
  799. var result = default(T);
  800. try
  801. {
  802. result = GetResult();
  803. }
  804. catch (Exception ex)
  805. {
  806. try { observer.OnError(ex); }
  807. finally { Dispose(); }
  808. return;
  809. }
  810. OnNext(result);
  811. }
  812. public void Done(int index)
  813. {
  814. isDone[index] = true;
  815. var allTrue = true;
  816. for (int i = 0; i < length; i++)
  817. {
  818. if (!isDone[i])
  819. {
  820. allTrue = false;
  821. break;
  822. }
  823. }
  824. if (allTrue)
  825. {
  826. try { observer.OnCompleted(); }
  827. finally { Dispose(); }
  828. }
  829. }
  830. public void Fail(Exception error)
  831. {
  832. try { observer.OnError(error); }
  833. finally { Dispose(); }
  834. }
  835. }
  836. // nth
  837. internal class ZipObserver<T> : IObserver<T>
  838. {
  839. readonly object gate;
  840. readonly IZipObservable parent;
  841. readonly int index;
  842. readonly Queue<T> queue;
  843. public ZipObserver(object gate, IZipObservable parent, int index, Queue<T> queue)
  844. {
  845. this.gate = gate;
  846. this.parent = parent;
  847. this.index = index;
  848. this.queue = queue;
  849. }
  850. public void OnNext(T value)
  851. {
  852. lock (gate)
  853. {
  854. queue.Enqueue(value);
  855. parent.Dequeue(index);
  856. }
  857. }
  858. public void OnError(Exception error)
  859. {
  860. lock (gate)
  861. {
  862. parent.Fail(error);
  863. }
  864. }
  865. public void OnCompleted()
  866. {
  867. lock (gate)
  868. {
  869. parent.Done(index);
  870. }
  871. }
  872. }
  873. }