JoinTest.cs 50 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112
  1. // Licensed to the .NET Foundation under one or more agreements.
  2. // The .NET Foundation licenses this file to you under the Apache 2.0 License.
  3. // See the LICENSE file in the project root for more information.
  4. using System;
  5. using System.Collections.Generic;
  6. using System.Linq;
  7. using System.Reactive;
  8. using System.Reactive.Linq;
  9. using Microsoft.Reactive.Testing;
  10. using ReactiveTests.Dummies;
  11. using Xunit;
  12. namespace ReactiveTests.Tests
  13. {
  14. public class JoinTest : ReactiveTest
  15. {
  16. [Fact]
  17. public void JoinOp_ArgumentChecking()
  18. {
  19. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Join(null, DummyObservable<int>.Instance, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, int, int>.Instance));
  20. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Join(DummyObservable<int>.Instance, null, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, int, int>.Instance));
  21. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Join(DummyObservable<int>.Instance, DummyObservable<int>.Instance, default(Func<int, IObservable<int>>), DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, int, int>.Instance));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Join(DummyObservable<int>.Instance, DummyObservable<int>.Instance, DummyFunc<int, IObservable<int>>.Instance, default(Func<int, IObservable<int>>), DummyFunc<int, int, int>.Instance));
  23. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Join(DummyObservable<int>.Instance, DummyObservable<int>.Instance, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, IObservable<int>>.Instance, default(Func<int, int, int>)));
  24. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Join(DummyObservable<int>.Instance, DummyObservable<int>.Instance, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, IObservable<int>>.Instance, DummyFunc<int, int, int>.Instance).Subscribe(null));
  25. }
  26. [Fact]
  27. public void JoinOp_Normal_I()
  28. {
  29. var scheduler = new TestScheduler();
  30. var xs = scheduler.CreateHotObservable(
  31. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  32. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  33. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  34. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  35. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  36. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  37. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  38. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  39. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  40. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  41. OnCompleted<TimeInterval<int>>(900)
  42. );
  43. var ys = scheduler.CreateHotObservable(
  44. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  45. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  46. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  47. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  48. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  49. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  50. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  51. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  52. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  53. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  54. OnCompleted<TimeInterval<string>>(800)
  55. );
  56. var xsd = new List<ITestableObservable<long>>();
  57. var ysd = new List<ITestableObservable<long>>();
  58. var res = scheduler.Start(() =>
  59. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  60. );
  61. res.Messages.AssertEqual(
  62. OnNext(215, "0hat"),
  63. OnNext(217, "0bat"),
  64. OnNext(219, "1hat"),
  65. OnNext(300, "3wag"),
  66. OnNext(300, "3pig"),
  67. OnNext(305, "3cup"),
  68. OnNext(310, "4wag"),
  69. OnNext(310, "4pig"),
  70. OnNext(310, "4cup"),
  71. OnNext(702, "6tin"),
  72. OnNext(710, "7tin"),
  73. OnNext(712, "6man"),
  74. OnNext(712, "7man"),
  75. OnNext(720, "8tin"),
  76. OnNext(720, "8man"),
  77. OnNext(722, "6rat"),
  78. OnNext(722, "7rat"),
  79. OnNext(722, "8rat"),
  80. OnNext(732, "7wig"),
  81. OnNext(732, "8wig"),
  82. OnNext(830, "9rat"),
  83. OnCompleted<string>(900)
  84. );
  85. xs.Subscriptions.AssertEqual(
  86. Subscribe(200, 900)
  87. );
  88. #if !NO_PERF // BREAKING CHANGE v2 > v1.x -> More aggressive disposal behavior
  89. ys.Subscriptions.AssertEqual(
  90. Subscribe(200, 800)
  91. );
  92. #else
  93. ys.Subscriptions.AssertEqual(
  94. Subscribe(200, 900)
  95. );
  96. #endif
  97. AssertDurations(xs, xsd, 900);
  98. AssertDurations(ys, ysd, 900);
  99. }
  100. [Fact]
  101. public void JoinOp_Normal_II()
  102. {
  103. var scheduler = new TestScheduler();
  104. var xs = scheduler.CreateHotObservable(
  105. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  106. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  107. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  108. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  109. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  110. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  111. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  112. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(200))),
  113. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  114. OnCompleted<TimeInterval<int>>(721)
  115. );
  116. var ys = scheduler.CreateHotObservable(
  117. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  118. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  119. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  120. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  121. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  122. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  123. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  124. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  125. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  126. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  127. OnCompleted<TimeInterval<string>>(990)
  128. );
  129. var xsd = new List<ITestableObservable<long>>();
  130. var ysd = new List<ITestableObservable<long>>();
  131. var res = scheduler.Start(() =>
  132. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  133. );
  134. res.Messages.AssertEqual(
  135. OnNext(215, "0hat"),
  136. OnNext(217, "0bat"),
  137. OnNext(219, "1hat"),
  138. OnNext(300, "3wag"),
  139. OnNext(300, "3pig"),
  140. OnNext(305, "3cup"),
  141. OnNext(310, "4wag"),
  142. OnNext(310, "4pig"),
  143. OnNext(310, "4cup"),
  144. OnNext(702, "6tin"),
  145. OnNext(710, "7tin"),
  146. OnNext(712, "6man"),
  147. OnNext(712, "7man"),
  148. OnNext(720, "8tin"),
  149. OnNext(720, "8man"),
  150. OnNext(722, "6rat"),
  151. OnNext(722, "7rat"),
  152. OnNext(722, "8rat"),
  153. OnNext(732, "7wig"),
  154. OnNext(732, "8wig"),
  155. OnCompleted<string>(910)
  156. );
  157. #if !NO_PERF // BREAKING CHANGE v2 > v1.x -> More aggressive disposal behavior
  158. xs.Subscriptions.AssertEqual(
  159. Subscribe(200, 721)
  160. );
  161. #else
  162. xs.Subscriptions.AssertEqual(
  163. Subscribe(200, 910)
  164. );
  165. #endif
  166. ys.Subscriptions.AssertEqual(
  167. Subscribe(200, 910)
  168. );
  169. AssertDurations(xs, xsd, 910);
  170. AssertDurations(ys, ysd, 910);
  171. }
  172. [Fact]
  173. public void JoinOp_Normal_III()
  174. {
  175. var scheduler = new TestScheduler();
  176. var xs = scheduler.CreateHotObservable(
  177. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  178. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  179. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  180. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  181. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  182. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  183. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  184. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  185. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  186. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  187. OnCompleted<TimeInterval<int>>(900)
  188. );
  189. var ys = scheduler.CreateHotObservable(
  190. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  191. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  192. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  193. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  194. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  195. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  196. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  197. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  198. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  199. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  200. OnCompleted<TimeInterval<string>>(800)
  201. );
  202. var xsd = new List<ITestableObservable<long>>();
  203. var ysd = new List<ITestableObservable<long>>();
  204. var res = scheduler.Start(() =>
  205. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler).Where(_ => false), y => NewTimer(ysd, y.Interval, scheduler).Where(_ => false), (x, y) => x.Value + y.Value)
  206. );
  207. res.Messages.AssertEqual(
  208. OnNext(215, "0hat"),
  209. OnNext(217, "0bat"),
  210. OnNext(219, "1hat"),
  211. OnNext(300, "3wag"),
  212. OnNext(300, "3pig"),
  213. OnNext(305, "3cup"),
  214. OnNext(310, "4wag"),
  215. OnNext(310, "4pig"),
  216. OnNext(310, "4cup"),
  217. OnNext(702, "6tin"),
  218. OnNext(710, "7tin"),
  219. OnNext(712, "6man"),
  220. OnNext(712, "7man"),
  221. OnNext(720, "8tin"),
  222. OnNext(720, "8man"),
  223. OnNext(722, "6rat"),
  224. OnNext(722, "7rat"),
  225. OnNext(722, "8rat"),
  226. OnNext(732, "7wig"),
  227. OnNext(732, "8wig"),
  228. OnNext(830, "9rat"),
  229. OnCompleted<string>(900)
  230. );
  231. xs.Subscriptions.AssertEqual(
  232. Subscribe(200, 900)
  233. );
  234. #if !NO_PERF // BREAKING CHANGE v2 > v1.x -> More aggressive disposal behavior
  235. ys.Subscriptions.AssertEqual(
  236. Subscribe(200, 800)
  237. );
  238. #else
  239. ys.Subscriptions.AssertEqual(
  240. Subscribe(200, 900)
  241. );
  242. #endif
  243. AssertDurations(xs, xsd, 900);
  244. AssertDurations(ys, ysd, 900);
  245. }
  246. [Fact]
  247. public void JoinOp_Normal_IV()
  248. {
  249. var scheduler = new TestScheduler();
  250. var xs = scheduler.CreateHotObservable(
  251. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  252. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  253. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  254. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  255. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  256. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  257. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  258. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(200))),
  259. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  260. OnCompleted<TimeInterval<int>>(990)
  261. );
  262. var ys = scheduler.CreateHotObservable(
  263. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  264. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  265. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  266. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  267. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  268. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  269. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  270. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  271. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  272. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  273. OnCompleted<TimeInterval<string>>(980)
  274. );
  275. var xsd = new List<ITestableObservable<long>>();
  276. var ysd = new List<ITestableObservable<long>>();
  277. var res = scheduler.Start(() =>
  278. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  279. );
  280. res.Messages.AssertEqual(
  281. OnNext(215, "0hat"),
  282. OnNext(217, "0bat"),
  283. OnNext(219, "1hat"),
  284. OnNext(300, "3wag"),
  285. OnNext(300, "3pig"),
  286. OnNext(305, "3cup"),
  287. OnNext(310, "4wag"),
  288. OnNext(310, "4pig"),
  289. OnNext(310, "4cup"),
  290. OnNext(702, "6tin"),
  291. OnNext(710, "7tin"),
  292. OnNext(712, "6man"),
  293. OnNext(712, "7man"),
  294. OnNext(720, "8tin"),
  295. OnNext(720, "8man"),
  296. OnNext(722, "6rat"),
  297. OnNext(722, "7rat"),
  298. OnNext(722, "8rat"),
  299. OnNext(732, "7wig"),
  300. OnNext(732, "8wig"),
  301. OnCompleted<string>(980)
  302. );
  303. xs.Subscriptions.AssertEqual(
  304. Subscribe(200, 980)
  305. );
  306. ys.Subscriptions.AssertEqual(
  307. Subscribe(200, 980)
  308. );
  309. AssertDurations(xs, xsd, 980);
  310. AssertDurations(ys, ysd, 980);
  311. }
  312. [Fact]
  313. public void JoinOp_Normal_V()
  314. {
  315. var scheduler = new TestScheduler();
  316. var xs = scheduler.CreateHotObservable(
  317. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  318. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  319. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  320. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  321. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  322. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  323. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  324. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(200))),
  325. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  326. OnCompleted<TimeInterval<int>>(990)
  327. );
  328. var ys = scheduler.CreateHotObservable(
  329. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  330. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  331. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  332. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  333. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  334. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  335. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  336. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  337. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  338. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  339. OnCompleted<TimeInterval<string>>(900)
  340. );
  341. var xsd = new List<ITestableObservable<long>>();
  342. var ysd = new List<ITestableObservable<long>>();
  343. var res = scheduler.Start(() =>
  344. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  345. );
  346. res.Messages.AssertEqual(
  347. OnNext(215, "0hat"),
  348. OnNext(217, "0bat"),
  349. OnNext(219, "1hat"),
  350. OnNext(300, "3wag"),
  351. OnNext(300, "3pig"),
  352. OnNext(305, "3cup"),
  353. OnNext(310, "4wag"),
  354. OnNext(310, "4pig"),
  355. OnNext(310, "4cup"),
  356. OnNext(702, "6tin"),
  357. OnNext(710, "7tin"),
  358. OnNext(712, "6man"),
  359. OnNext(712, "7man"),
  360. OnNext(720, "8tin"),
  361. OnNext(720, "8man"),
  362. OnNext(722, "6rat"),
  363. OnNext(722, "7rat"),
  364. OnNext(722, "8rat"),
  365. OnNext(732, "7wig"),
  366. OnNext(732, "8wig"),
  367. OnCompleted<string>(922)
  368. );
  369. xs.Subscriptions.AssertEqual(
  370. Subscribe(200, 922)
  371. );
  372. #if !NO_PERF // BREAKING CHANGE v2 > v1.x -> More aggressive disposal behavior
  373. ys.Subscriptions.AssertEqual(
  374. Subscribe(200, 900)
  375. );
  376. #else
  377. ys.Subscriptions.AssertEqual(
  378. Subscribe(200, 922)
  379. );
  380. #endif
  381. AssertDurations(xs, xsd, 922);
  382. AssertDurations(ys, ysd, 922);
  383. }
  384. [Fact]
  385. public void JoinOp_Normal_VI()
  386. {
  387. var scheduler = new TestScheduler();
  388. var xs = scheduler.CreateHotObservable(
  389. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  390. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  391. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  392. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  393. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  394. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  395. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  396. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(30))),
  397. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(200))),
  398. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  399. OnCompleted<TimeInterval<int>>(850)
  400. );
  401. var ys = scheduler.CreateHotObservable(
  402. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  403. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  404. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  405. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  406. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  407. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  408. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  409. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  410. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(20))),
  411. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  412. OnCompleted<TimeInterval<string>>(900)
  413. );
  414. var xsd = new List<ITestableObservable<long>>();
  415. var ysd = new List<ITestableObservable<long>>();
  416. var res = scheduler.Start(() =>
  417. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  418. );
  419. res.Messages.AssertEqual(
  420. OnNext(215, "0hat"),
  421. OnNext(217, "0bat"),
  422. OnNext(219, "1hat"),
  423. OnNext(300, "3wag"),
  424. OnNext(300, "3pig"),
  425. OnNext(305, "3cup"),
  426. OnNext(310, "4wag"),
  427. OnNext(310, "4pig"),
  428. OnNext(310, "4cup"),
  429. OnNext(702, "6tin"),
  430. OnNext(710, "7tin"),
  431. OnNext(712, "6man"),
  432. OnNext(712, "7man"),
  433. OnNext(720, "8tin"),
  434. OnNext(720, "8man"),
  435. OnNext(722, "6rat"),
  436. OnNext(722, "7rat"),
  437. OnNext(722, "8rat"),
  438. OnNext(732, "7wig"),
  439. OnNext(732, "8wig"),
  440. OnCompleted<string>(900)
  441. );
  442. #if !NO_PERF // BREAKING CHANGE v2 > v1.x -> More aggressive disposal behavior
  443. xs.Subscriptions.AssertEqual(
  444. Subscribe(200, 850)
  445. );
  446. #else
  447. xs.Subscriptions.AssertEqual(
  448. Subscribe(200, 900)
  449. );
  450. #endif
  451. ys.Subscriptions.AssertEqual(
  452. Subscribe(200, 900)
  453. );
  454. AssertDurations(xs, xsd, 900);
  455. AssertDurations(ys, ysd, 900);
  456. }
  457. [Fact]
  458. public void JoinOp_Normal_VII()
  459. {
  460. var scheduler = new TestScheduler();
  461. var xs = scheduler.CreateHotObservable(
  462. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  463. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  464. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  465. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  466. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  467. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  468. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  469. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  470. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  471. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  472. OnCompleted<TimeInterval<int>>(900)
  473. );
  474. var ys = scheduler.CreateHotObservable(
  475. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  476. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  477. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  478. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  479. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  480. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  481. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  482. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  483. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  484. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  485. OnCompleted<TimeInterval<string>>(800)
  486. );
  487. var xsd = new List<ITestableObservable<long>>();
  488. var ysd = new List<ITestableObservable<long>>();
  489. var res = scheduler.Start(() =>
  490. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value),
  491. 713
  492. );
  493. res.Messages.AssertEqual(
  494. OnNext(215, "0hat"),
  495. OnNext(217, "0bat"),
  496. OnNext(219, "1hat"),
  497. OnNext(300, "3wag"),
  498. OnNext(300, "3pig"),
  499. OnNext(305, "3cup"),
  500. OnNext(310, "4wag"),
  501. OnNext(310, "4pig"),
  502. OnNext(310, "4cup"),
  503. OnNext(702, "6tin"),
  504. OnNext(710, "7tin"),
  505. OnNext(712, "6man"),
  506. OnNext(712, "7man")
  507. );
  508. xs.Subscriptions.AssertEqual(
  509. Subscribe(200, 713)
  510. );
  511. ys.Subscriptions.AssertEqual(
  512. Subscribe(200, 713)
  513. );
  514. AssertDurations(xs, xsd, 713);
  515. AssertDurations(ys, ysd, 713);
  516. }
  517. [Fact]
  518. public void JoinOp_Error_I()
  519. {
  520. var scheduler = new TestScheduler();
  521. var ex = new Exception();
  522. var xs = scheduler.CreateHotObservable(
  523. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  524. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  525. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  526. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  527. OnError<TimeInterval<int>>(310, ex)
  528. );
  529. var ys = scheduler.CreateHotObservable(
  530. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  531. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  532. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  533. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  534. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  535. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  536. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  537. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  538. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  539. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  540. OnCompleted<TimeInterval<string>>(800)
  541. );
  542. var xsd = new List<ITestableObservable<long>>();
  543. var ysd = new List<ITestableObservable<long>>();
  544. var res = scheduler.Start(() =>
  545. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  546. );
  547. res.Messages.AssertEqual(
  548. OnNext(215, "0hat"),
  549. OnNext(217, "0bat"),
  550. OnNext(219, "1hat"),
  551. OnNext(300, "3wag"),
  552. OnNext(300, "3pig"),
  553. OnNext(305, "3cup"),
  554. OnError<string>(310, ex)
  555. );
  556. xs.Subscriptions.AssertEqual(
  557. Subscribe(200, 310)
  558. );
  559. ys.Subscriptions.AssertEqual(
  560. Subscribe(200, 310)
  561. );
  562. AssertDurations(xs, xsd, 310);
  563. AssertDurations(ys, ysd, 310);
  564. }
  565. [Fact]
  566. public void JoinOp_Error_II()
  567. {
  568. var scheduler = new TestScheduler();
  569. var ex = new Exception();
  570. var xs = scheduler.CreateHotObservable(
  571. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  572. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  573. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  574. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  575. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  576. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  577. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  578. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  579. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  580. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  581. OnCompleted<TimeInterval<int>>(900)
  582. );
  583. var ys = scheduler.CreateHotObservable(
  584. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  585. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  586. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  587. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  588. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  589. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  590. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  591. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  592. OnError<TimeInterval<string>>(722, ex)
  593. );
  594. var xsd = new List<ITestableObservable<long>>();
  595. var ysd = new List<ITestableObservable<long>>();
  596. var res = scheduler.Start(() =>
  597. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  598. );
  599. res.Messages.AssertEqual(
  600. OnNext(215, "0hat"),
  601. OnNext(217, "0bat"),
  602. OnNext(219, "1hat"),
  603. OnNext(300, "3wag"),
  604. OnNext(300, "3pig"),
  605. OnNext(305, "3cup"),
  606. OnNext(310, "4wag"),
  607. OnNext(310, "4pig"),
  608. OnNext(310, "4cup"),
  609. OnNext(702, "6tin"),
  610. OnNext(710, "7tin"),
  611. OnNext(712, "6man"),
  612. OnNext(712, "7man"),
  613. OnNext(720, "8tin"),
  614. OnNext(720, "8man"),
  615. OnError<string>(722, ex)
  616. );
  617. xs.Subscriptions.AssertEqual(
  618. Subscribe(200, 722)
  619. );
  620. ys.Subscriptions.AssertEqual(
  621. Subscribe(200, 722)
  622. );
  623. AssertDurations(xs, xsd, 722);
  624. AssertDurations(ys, ysd, 722);
  625. }
  626. [Fact]
  627. public void JoinOp_Error_III()
  628. {
  629. var scheduler = new TestScheduler();
  630. var xs = scheduler.CreateHotObservable(
  631. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  632. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  633. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  634. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  635. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  636. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  637. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  638. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  639. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  640. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  641. OnCompleted<TimeInterval<int>>(900)
  642. );
  643. var ys = scheduler.CreateHotObservable(
  644. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  645. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  646. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  647. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  648. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  649. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  650. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  651. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  652. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  653. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  654. OnCompleted<TimeInterval<string>>(800)
  655. );
  656. var ex = new Exception();
  657. var xsd = new List<ITestableObservable<long>>();
  658. var ysd = new List<ITestableObservable<long>>();
  659. var res = scheduler.Start(() =>
  660. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler).SelectMany(x.Value == 6 ? Observable.Throw<long>(ex) : Observable.Empty<long>()), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  661. );
  662. res.Messages.AssertEqual(
  663. OnNext(215, "0hat"),
  664. OnNext(217, "0bat"),
  665. OnNext(219, "1hat"),
  666. OnNext(300, "3wag"),
  667. OnNext(300, "3pig"),
  668. OnNext(305, "3cup"),
  669. OnNext(310, "4wag"),
  670. OnNext(310, "4pig"),
  671. OnNext(310, "4cup"),
  672. OnNext(702, "6tin"),
  673. OnNext(710, "7tin"),
  674. OnNext(712, "6man"),
  675. OnNext(712, "7man"),
  676. OnNext(720, "8tin"),
  677. OnNext(720, "8man"),
  678. OnNext(722, "6rat"),
  679. OnNext(722, "7rat"),
  680. OnNext(722, "8rat"),
  681. OnError<string>(725, ex)
  682. );
  683. xs.Subscriptions.AssertEqual(
  684. Subscribe(200, 725)
  685. );
  686. ys.Subscriptions.AssertEqual(
  687. Subscribe(200, 725)
  688. );
  689. AssertDurations(xs, xsd, 725);
  690. AssertDurations(ys, ysd, 725);
  691. }
  692. [Fact]
  693. public void JoinOp_Error_IV()
  694. {
  695. var scheduler = new TestScheduler();
  696. var xs = scheduler.CreateHotObservable(
  697. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  698. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  699. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  700. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  701. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  702. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  703. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  704. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  705. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  706. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  707. OnCompleted<TimeInterval<int>>(900)
  708. );
  709. var ys = scheduler.CreateHotObservable(
  710. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  711. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  712. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  713. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  714. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  715. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  716. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(19))),
  717. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  718. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  719. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  720. OnCompleted<TimeInterval<string>>(800)
  721. );
  722. var ex = new Exception();
  723. var xsd = new List<ITestableObservable<long>>();
  724. var ysd = new List<ITestableObservable<long>>();
  725. var res = scheduler.Start(() =>
  726. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler).SelectMany(y.Value == "tin" ? Observable.Throw<long>(ex) : Observable.Empty<long>()), (x, y) => x.Value + y.Value)
  727. );
  728. res.Messages.AssertEqual(
  729. OnNext(215, "0hat"),
  730. OnNext(217, "0bat"),
  731. OnNext(219, "1hat"),
  732. OnNext(300, "3wag"),
  733. OnNext(300, "3pig"),
  734. OnNext(305, "3cup"),
  735. OnNext(310, "4wag"),
  736. OnNext(310, "4pig"),
  737. OnNext(310, "4cup"),
  738. OnNext(702, "6tin"),
  739. OnNext(710, "7tin"),
  740. OnNext(712, "6man"),
  741. OnNext(712, "7man"),
  742. OnNext(720, "8tin"),
  743. OnNext(720, "8man"),
  744. OnError<string>(721, ex)
  745. );
  746. xs.Subscriptions.AssertEqual(
  747. Subscribe(200, 721)
  748. );
  749. ys.Subscriptions.AssertEqual(
  750. Subscribe(200, 721)
  751. );
  752. AssertDurations(xs, xsd, 721);
  753. AssertDurations(ys, ysd, 721);
  754. }
  755. [Fact]
  756. public void JoinOp_Error_V()
  757. {
  758. var scheduler = new TestScheduler();
  759. var xs = scheduler.CreateHotObservable(
  760. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  761. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  762. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  763. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  764. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  765. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  766. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  767. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  768. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  769. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  770. OnCompleted<TimeInterval<int>>(900)
  771. );
  772. var ys = scheduler.CreateHotObservable(
  773. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  774. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  775. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  776. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  777. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  778. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  779. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  780. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  781. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  782. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  783. OnCompleted<TimeInterval<string>>(800)
  784. );
  785. var ex = new Exception();
  786. var ysd = new List<ITestableObservable<long>>();
  787. var res = scheduler.Start(() =>
  788. xs.Join(ys, x => { if (x.Value >= 0) { throw ex; } return Observable.Empty<long>(); }, y => NewTimer(ysd, y.Interval, scheduler), (x, y) => x.Value + y.Value)
  789. );
  790. res.Messages.AssertEqual(
  791. OnError<string>(210, ex)
  792. );
  793. xs.Subscriptions.AssertEqual(
  794. Subscribe(200, 210)
  795. );
  796. ys.Subscriptions.AssertEqual(
  797. Subscribe(200, 210)
  798. );
  799. AssertDurations(ys, ysd, 210);
  800. }
  801. [Fact]
  802. public void JoinOp_Error_VI()
  803. {
  804. var scheduler = new TestScheduler();
  805. var xs = scheduler.CreateHotObservable(
  806. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  807. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  808. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  809. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  810. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  811. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  812. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  813. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  814. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  815. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  816. OnCompleted<TimeInterval<int>>(900)
  817. );
  818. var ys = scheduler.CreateHotObservable(
  819. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  820. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  821. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  822. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  823. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  824. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  825. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  826. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  827. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  828. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  829. OnCompleted<TimeInterval<string>>(800)
  830. );
  831. var ex = new Exception();
  832. var xsd = new List<ITestableObservable<long>>();
  833. var res = scheduler.Start(() =>
  834. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => { if (y.Value.Length >= 0) { throw ex; } return Observable.Empty<long>(); }, (x, y) => x.Value + y.Value)
  835. );
  836. res.Messages.AssertEqual(
  837. OnError<string>(215, ex)
  838. );
  839. xs.Subscriptions.AssertEqual(
  840. Subscribe(200, 215)
  841. );
  842. ys.Subscriptions.AssertEqual(
  843. Subscribe(200, 215)
  844. );
  845. AssertDurations(xs, xsd, 215);
  846. }
  847. [Fact]
  848. public void JoinOp_Error_VII()
  849. {
  850. var scheduler = new TestScheduler();
  851. var xs = scheduler.CreateHotObservable(
  852. OnNext(215, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  853. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  854. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  855. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  856. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  857. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  858. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  859. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  860. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  861. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  862. OnCompleted<TimeInterval<int>>(900)
  863. );
  864. var ys = scheduler.CreateHotObservable(
  865. OnNext(210, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  866. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  867. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  868. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  869. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  870. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  871. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  872. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  873. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  874. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  875. OnCompleted<TimeInterval<string>>(800)
  876. );
  877. var ex = new Exception();
  878. var xsd = new List<ITestableObservable<long>>();
  879. var ysd = new List<ITestableObservable<long>>();
  880. var res = scheduler.Start(() =>
  881. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => { if (x.Value >= 0) { throw ex; } return x.Value + y.Value; })
  882. );
  883. res.Messages.AssertEqual(
  884. OnError<string>(215, ex)
  885. );
  886. xs.Subscriptions.AssertEqual(
  887. Subscribe(200, 215)
  888. );
  889. ys.Subscriptions.AssertEqual(
  890. Subscribe(200, 215)
  891. );
  892. AssertDurations(xs, xsd, 215);
  893. AssertDurations(ys, ysd, 215);
  894. }
  895. [Fact]
  896. public void JoinOp_Error_VIII()
  897. {
  898. var scheduler = new TestScheduler();
  899. var xs = scheduler.CreateHotObservable(
  900. OnNext(210, new TimeInterval<int>(0, TimeSpan.FromTicks(10))),
  901. OnNext(219, new TimeInterval<int>(1, TimeSpan.FromTicks(5))),
  902. OnNext(240, new TimeInterval<int>(2, TimeSpan.FromTicks(10))),
  903. OnNext(300, new TimeInterval<int>(3, TimeSpan.FromTicks(100))),
  904. OnNext(310, new TimeInterval<int>(4, TimeSpan.FromTicks(80))),
  905. OnNext(500, new TimeInterval<int>(5, TimeSpan.FromTicks(90))),
  906. OnNext(700, new TimeInterval<int>(6, TimeSpan.FromTicks(25))),
  907. OnNext(710, new TimeInterval<int>(7, TimeSpan.FromTicks(300))),
  908. OnNext(720, new TimeInterval<int>(8, TimeSpan.FromTicks(100))),
  909. OnNext(830, new TimeInterval<int>(9, TimeSpan.FromTicks(10))),
  910. OnCompleted<TimeInterval<int>>(900)
  911. );
  912. var ys = scheduler.CreateHotObservable(
  913. OnNext(215, new TimeInterval<string>("hat", TimeSpan.FromTicks(20))),
  914. OnNext(217, new TimeInterval<string>("bat", TimeSpan.FromTicks(1))),
  915. OnNext(290, new TimeInterval<string>("wag", TimeSpan.FromTicks(200))),
  916. OnNext(300, new TimeInterval<string>("pig", TimeSpan.FromTicks(10))),
  917. OnNext(305, new TimeInterval<string>("cup", TimeSpan.FromTicks(50))),
  918. OnNext(600, new TimeInterval<string>("yak", TimeSpan.FromTicks(90))),
  919. OnNext(702, new TimeInterval<string>("tin", TimeSpan.FromTicks(20))),
  920. OnNext(712, new TimeInterval<string>("man", TimeSpan.FromTicks(10))),
  921. OnNext(722, new TimeInterval<string>("rat", TimeSpan.FromTicks(200))),
  922. OnNext(732, new TimeInterval<string>("wig", TimeSpan.FromTicks(5))),
  923. OnCompleted<TimeInterval<string>>(800)
  924. );
  925. var ex = new Exception();
  926. var xsd = new List<ITestableObservable<long>>();
  927. var ysd = new List<ITestableObservable<long>>();
  928. var res = scheduler.Start(() =>
  929. xs.Join(ys, x => NewTimer(xsd, x.Interval, scheduler), y => NewTimer(ysd, y.Interval, scheduler), (x, y) => { if (x.Value >= 0) { throw ex; } return x.Value + y.Value; })
  930. );
  931. res.Messages.AssertEqual(
  932. OnError<string>(215, ex)
  933. );
  934. xs.Subscriptions.AssertEqual(
  935. Subscribe(200, 215)
  936. );
  937. ys.Subscriptions.AssertEqual(
  938. Subscribe(200, 215)
  939. );
  940. AssertDurations(xs, xsd, 215);
  941. AssertDurations(ys, ysd, 215);
  942. }
  943. private ITestableObservable<long> NewTimer(List<ITestableObservable<long>> l, TimeSpan t, TestScheduler scheduler)
  944. {
  945. var timer = scheduler.CreateColdObservable(OnNext(t.Ticks, 0L), OnCompleted<long>(t.Ticks));
  946. l.Add(timer);
  947. return timer;
  948. }
  949. private void AssertDurations<T, U>(ITestableObservable<TimeInterval<T>> xs, List<ITestableObservable<U>> xsd, long lastEnd)
  950. {
  951. Assert.Equal(xs.Messages.Where(x => x.Value.Kind == NotificationKind.OnNext && x.Time <= lastEnd).Count(), xsd.Count);
  952. foreach (var pair in xs.Messages.Zip(xsd, (x, y) => new { Item1 = x, Item2 = y }))
  953. {
  954. var start = pair.Item1.Time;
  955. var end = Math.Min(start + pair.Item1.Value.Value.Interval.Ticks, lastEnd);
  956. pair.Item2.Subscriptions.AssertEqual(
  957. Subscribe(start, end)
  958. );
  959. }
  960. }
  961. }
  962. }