DelayTest.cs 51 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846
  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.Concurrency;
  9. using System.Reactive.Disposables;
  10. using System.Reactive.Linq;
  11. using System.Reactive.Subjects;
  12. using System.Threading;
  13. using System.Threading.Tasks;
  14. using Microsoft.Reactive.Testing;
  15. using ReactiveTests.Dummies;
  16. using Xunit;
  17. namespace ReactiveTests.Tests
  18. {
  19. public class DelayTest : ReactiveTest
  20. {
  21. [Fact]
  22. public void Delay_ArgumentChecking()
  23. {
  24. var scheduler = new TestScheduler();
  25. var someObservable = Observable.Empty<int>();
  26. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(default(IObservable<int>), DateTimeOffset.Now));
  27. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(default(IObservable<int>), TimeSpan.Zero));
  28. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Observable.Delay(someObservable, TimeSpan.FromSeconds(-1)));
  29. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(default(IObservable<int>), DateTimeOffset.Now, scheduler));
  30. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(default(IObservable<int>), TimeSpan.Zero, scheduler));
  31. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(someObservable, DateTimeOffset.Now, null));
  32. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(someObservable, TimeSpan.Zero, null));
  33. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Observable.Delay(someObservable, TimeSpan.FromSeconds(-1), scheduler));
  34. }
  35. [Fact]
  36. public void Delay_TimeSpan_Simple1()
  37. {
  38. Delay_TimeSpan_Simple1_Impl(false);
  39. }
  40. [Fact]
  41. public void Delay_TimeSpan_Simple1_Stopwatch()
  42. {
  43. Delay_TimeSpan_Simple1_Impl(true);
  44. }
  45. private void Delay_TimeSpan_Simple1_Impl(bool useStopwatch)
  46. {
  47. var scheduler = new TestScheduler();
  48. var xs = scheduler.CreateHotObservable(
  49. OnNext(150, 1),
  50. OnNext(250, 2),
  51. OnNext(350, 3),
  52. OnNext(450, 4),
  53. OnCompleted<int>(550)
  54. );
  55. var res = scheduler.Start(() =>
  56. xs.Delay(TimeSpan.FromTicks(100), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  57. );
  58. res.Messages.AssertEqual(
  59. OnNext(350, 2),
  60. OnNext(450, 3),
  61. OnNext(550, 4),
  62. OnCompleted<int>(650)
  63. );
  64. xs.Subscriptions.AssertEqual(
  65. Subscribe(200, 550)
  66. );
  67. }
  68. [Fact]
  69. public void Delay_DateTimeOffset_Simple1()
  70. {
  71. Delay_DateTimeOffset_Simple1_Impl(false);
  72. }
  73. [Fact]
  74. public void Delay_DateTimeOffset_Simple1_Stopwatch()
  75. {
  76. Delay_DateTimeOffset_Simple1_Impl(true);
  77. }
  78. private void Delay_DateTimeOffset_Simple1_Impl(bool useStopwatch)
  79. {
  80. var scheduler = new TestScheduler();
  81. var xs = scheduler.CreateHotObservable(
  82. OnNext(150, 1),
  83. OnNext(250, 2),
  84. OnNext(350, 3),
  85. OnNext(450, 4),
  86. OnCompleted<int>(550)
  87. );
  88. var res = scheduler.Start(() =>
  89. xs.Delay(new DateTimeOffset(300, TimeSpan.Zero), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  90. );
  91. res.Messages.AssertEqual(
  92. OnNext(350, 2),
  93. OnNext(450, 3),
  94. OnNext(550, 4),
  95. OnCompleted<int>(650)
  96. );
  97. xs.Subscriptions.AssertEqual(
  98. Subscribe(200, 550)
  99. );
  100. }
  101. [Fact]
  102. public void Delay_TimeSpan_Simple2()
  103. {
  104. Delay_TimeSpan_Simple2_Impl(false);
  105. }
  106. [Fact]
  107. public void Delay_TimeSpan_Simple2_Stopwatch()
  108. {
  109. Delay_TimeSpan_Simple2_Impl(true);
  110. }
  111. private void Delay_TimeSpan_Simple2_Impl(bool useStopwatch)
  112. {
  113. var scheduler = new TestScheduler();
  114. var xs = scheduler.CreateHotObservable(
  115. OnNext(150, 1),
  116. OnNext(250, 2),
  117. OnNext(350, 3),
  118. OnNext(450, 4),
  119. OnCompleted<int>(550)
  120. );
  121. var res = scheduler.Start(() =>
  122. xs.Delay(TimeSpan.FromTicks(50), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  123. );
  124. res.Messages.AssertEqual(
  125. OnNext(300, 2),
  126. OnNext(400, 3),
  127. OnNext(500, 4),
  128. OnCompleted<int>(600)
  129. );
  130. xs.Subscriptions.AssertEqual(
  131. Subscribe(200, 550)
  132. );
  133. }
  134. [Fact]
  135. public void Delay_DateTimeOffset_Simple2()
  136. {
  137. Delay_DateTimeOffset_Simple2_Impl(false);
  138. }
  139. [Fact]
  140. public void Delay_DateTimeOffset_Simple2_Stopwatch()
  141. {
  142. Delay_DateTimeOffset_Simple2_Impl(true);
  143. }
  144. private void Delay_DateTimeOffset_Simple2_Impl(bool useStopwatch)
  145. {
  146. var scheduler = new TestScheduler();
  147. var xs = scheduler.CreateHotObservable(
  148. OnNext(150, 1),
  149. OnNext(250, 2),
  150. OnNext(350, 3),
  151. OnNext(450, 4),
  152. OnCompleted<int>(550)
  153. );
  154. var res = scheduler.Start(() =>
  155. xs.Delay(new DateTimeOffset(250, TimeSpan.Zero), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  156. );
  157. res.Messages.AssertEqual(
  158. OnNext(300, 2),
  159. OnNext(400, 3),
  160. OnNext(500, 4),
  161. OnCompleted<int>(600)
  162. );
  163. xs.Subscriptions.AssertEqual(
  164. Subscribe(200, 550)
  165. );
  166. }
  167. [Fact]
  168. public void Delay_TimeSpan_Simple3()
  169. {
  170. Delay_TimeSpan_Simple3_Impl(false);
  171. }
  172. [Fact]
  173. public void Delay_TimeSpan_Simple3_Stopwatch()
  174. {
  175. Delay_TimeSpan_Simple3_Impl(true);
  176. }
  177. private void Delay_TimeSpan_Simple3_Impl(bool useStopwatch)
  178. {
  179. var scheduler = new TestScheduler();
  180. var xs = scheduler.CreateHotObservable(
  181. OnNext(150, 1),
  182. OnNext(250, 2),
  183. OnNext(350, 3),
  184. OnNext(450, 4),
  185. OnCompleted<int>(550)
  186. );
  187. var res = scheduler.Start(() =>
  188. xs.Delay(TimeSpan.FromTicks(150), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  189. );
  190. res.Messages.AssertEqual(
  191. OnNext(400, 2),
  192. OnNext(500, 3),
  193. OnNext(600, 4),
  194. OnCompleted<int>(700)
  195. );
  196. xs.Subscriptions.AssertEqual(
  197. Subscribe(200, 550)
  198. );
  199. }
  200. [Fact]
  201. public void Delay_DateTimeOffset_Simple3()
  202. {
  203. Delay_DateTimeOffset_Simple3_Impl(false);
  204. }
  205. [Fact]
  206. public void Delay_DateTimeOffset_Simple3_Stopwatch()
  207. {
  208. Delay_DateTimeOffset_Simple3_Impl(true);
  209. }
  210. private void Delay_DateTimeOffset_Simple3_Impl(bool useStopwatch)
  211. {
  212. var scheduler = new TestScheduler();
  213. var xs = scheduler.CreateHotObservable(
  214. OnNext(150, 1),
  215. OnNext(250, 2),
  216. OnNext(350, 3),
  217. OnNext(450, 4),
  218. OnCompleted<int>(550)
  219. );
  220. var res = scheduler.Start(() =>
  221. xs.Delay(new DateTimeOffset(350, TimeSpan.Zero), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  222. );
  223. res.Messages.AssertEqual(
  224. OnNext(400, 2),
  225. OnNext(500, 3),
  226. OnNext(600, 4),
  227. OnCompleted<int>(700)
  228. );
  229. xs.Subscriptions.AssertEqual(
  230. Subscribe(200, 550)
  231. );
  232. }
  233. [Fact]
  234. public void Delay_TimeSpan_Error1()
  235. {
  236. Delay_TimeSpan_Error1_Impl(false);
  237. }
  238. [Fact]
  239. public void Delay_TimeSpan_Error1_Stopwatch()
  240. {
  241. Delay_TimeSpan_Error1_Impl(true);
  242. }
  243. private void Delay_TimeSpan_Error1_Impl(bool useStopwatch)
  244. {
  245. var scheduler = new TestScheduler();
  246. var ex = new Exception();
  247. var xs = scheduler.CreateHotObservable(
  248. OnNext(150, 1),
  249. OnNext(250, 2),
  250. OnNext(350, 3),
  251. OnNext(450, 4),
  252. OnError<int>(550, ex)
  253. );
  254. var res = scheduler.Start(() =>
  255. xs.Delay(TimeSpan.FromTicks(50), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  256. );
  257. res.Messages.AssertEqual(
  258. OnNext(300, 2),
  259. OnNext(400, 3),
  260. OnNext(500, 4),
  261. OnError<int>(550, ex)
  262. );
  263. xs.Subscriptions.AssertEqual(
  264. Subscribe(200, 550)
  265. );
  266. }
  267. [Fact]
  268. public void Delay_DateTimeOffset_Error1()
  269. {
  270. Delay_DateTimeOffset_Error1_Impl(false);
  271. }
  272. [Fact]
  273. public void Delay_DateTimeOffset_Error1_Stopwatch()
  274. {
  275. Delay_DateTimeOffset_Error1_Impl(true);
  276. }
  277. private void Delay_DateTimeOffset_Error1_Impl(bool useStopwatch)
  278. {
  279. var scheduler = new TestScheduler();
  280. var ex = new Exception();
  281. var xs = scheduler.CreateHotObservable(
  282. OnNext(150, 1),
  283. OnNext(250, 2),
  284. OnNext(350, 3),
  285. OnNext(450, 4),
  286. OnError<int>(550, ex)
  287. );
  288. var res = scheduler.Start(() =>
  289. xs.Delay(new DateTimeOffset(250, TimeSpan.Zero), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  290. );
  291. res.Messages.AssertEqual(
  292. OnNext(300, 2),
  293. OnNext(400, 3),
  294. OnNext(500, 4),
  295. OnError<int>(550, ex)
  296. );
  297. xs.Subscriptions.AssertEqual(
  298. Subscribe(200, 550)
  299. );
  300. }
  301. [Fact]
  302. public void Delay_TimeSpan_Error2()
  303. {
  304. Delay_TimeSpan_Error2_Impl(false);
  305. }
  306. [Fact]
  307. public void Delay_TimeSpan_Error2_Stopwatch()
  308. {
  309. Delay_TimeSpan_Error2_Impl(true);
  310. }
  311. private void Delay_TimeSpan_Error2_Impl(bool useStopwatch)
  312. {
  313. var scheduler = new TestScheduler();
  314. var ex = new Exception();
  315. var xs = scheduler.CreateHotObservable(
  316. OnNext(150, 1),
  317. OnNext(250, 2),
  318. OnNext(350, 3),
  319. OnNext(450, 4),
  320. OnError<int>(550, ex)
  321. );
  322. var res = scheduler.Start(() =>
  323. xs.Delay(TimeSpan.FromTicks(150), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  324. );
  325. res.Messages.AssertEqual(
  326. OnNext(400, 2),
  327. OnNext(500, 3),
  328. OnError<int>(550, ex)
  329. );
  330. xs.Subscriptions.AssertEqual(
  331. Subscribe(200, 550)
  332. );
  333. }
  334. [Fact]
  335. public void Delay_DateTimeOffset_Error2()
  336. {
  337. Delay_DateTimeOffset_Error2_Impl(false);
  338. }
  339. [Fact]
  340. public void Delay_DateTimeOffset_Error2_Stopwatch()
  341. {
  342. Delay_DateTimeOffset_Error2_Impl(true);
  343. }
  344. private void Delay_DateTimeOffset_Error2_Impl(bool useStopwatch)
  345. {
  346. var scheduler = new TestScheduler();
  347. var ex = new Exception();
  348. var xs = scheduler.CreateHotObservable(
  349. OnNext(150, 1),
  350. OnNext(250, 2),
  351. OnNext(350, 3),
  352. OnNext(450, 4),
  353. OnError<int>(550, ex)
  354. );
  355. var res = scheduler.Start(() =>
  356. xs.Delay(new DateTimeOffset(350, TimeSpan.Zero), useStopwatch ? scheduler : scheduler.DisableOptimizations())
  357. );
  358. res.Messages.AssertEqual(
  359. OnNext(400, 2),
  360. OnNext(500, 3),
  361. OnError<int>(550, ex)
  362. );
  363. xs.Subscriptions.AssertEqual(
  364. Subscribe(200, 550)
  365. );
  366. }
  367. #if !NO_THREAD
  368. [Fact]
  369. public void Delay_TimeSpan_Real_Simple1()
  370. {
  371. Delay_TimeSpan_Real_Simple1_Impl(ThreadPoolScheduler.Instance.DisableOptimizations());
  372. }
  373. [Fact]
  374. public void Delay_TimeSpan_Real_Simple1_Stopwatch()
  375. {
  376. Delay_TimeSpan_Real_Simple1_Impl(ThreadPoolScheduler.Instance);
  377. }
  378. #endif
  379. private void Delay_TimeSpan_Real_Simple1_Impl(IScheduler scheduler)
  380. {
  381. var s = new Subject<int>();
  382. var res = s.Delay(TimeSpan.FromMilliseconds(10), scheduler);
  383. var lst = new List<int>();
  384. var e = new ManualResetEvent(false);
  385. res.Subscribe(lst.Add, () => e.Set());
  386. Task.Run(() =>
  387. {
  388. s.OnNext(1);
  389. s.OnNext(2);
  390. s.OnNext(3);
  391. s.OnCompleted();
  392. });
  393. e.WaitOne();
  394. Assert.True(new[] { 1, 2, 3 }.SequenceEqual(lst));
  395. }
  396. #if !NO_THREAD
  397. [Fact]
  398. public void Delay_TimeSpan_Real_Error1()
  399. {
  400. Delay_TimeSpan_Real_Error1_Impl(ThreadPoolScheduler.Instance.DisableOptimizations());
  401. }
  402. [Fact]
  403. public void Delay_TimeSpan_Real_Error1_Stopwatch()
  404. {
  405. Delay_TimeSpan_Real_Error1_Impl(ThreadPoolScheduler.Instance);
  406. }
  407. #endif
  408. private void Delay_TimeSpan_Real_Error1_Impl(IScheduler scheduler)
  409. {
  410. var ex = new Exception();
  411. var s = new Subject<int>();
  412. var res = s.Delay(TimeSpan.FromMilliseconds(10), scheduler);
  413. var e = new ManualResetEvent(false);
  414. var err = default(Exception);
  415. res.Subscribe(_ => { }, ex_ => { err = ex_; e.Set(); });
  416. Task.Run(() =>
  417. {
  418. s.OnNext(1);
  419. s.OnNext(2);
  420. s.OnNext(3);
  421. s.OnError(ex);
  422. });
  423. e.WaitOne();
  424. Assert.Same(ex, err);
  425. }
  426. #if !NO_THREAD
  427. [Fact]
  428. public void Delay_TimeSpan_Real_Error2()
  429. {
  430. Delay_TimeSpan_Real_Error2_Impl(ThreadPoolScheduler.Instance.DisableOptimizations());
  431. }
  432. [Fact]
  433. public void Delay_TimeSpan_Real_Error2_Stopwatch()
  434. {
  435. Delay_TimeSpan_Real_Error2_Impl(ThreadPoolScheduler.Instance);
  436. }
  437. #endif
  438. private void Delay_TimeSpan_Real_Error2_Impl(IScheduler scheduler)
  439. {
  440. var ex = new Exception();
  441. var s = new Subject<int>();
  442. var res = s.Delay(TimeSpan.FromMilliseconds(10), scheduler);
  443. var next = new ManualResetEvent(false);
  444. var e = new ManualResetEvent(false);
  445. var err = default(Exception);
  446. res.Subscribe(_ => { next.Set(); }, ex_ => { err = ex_; e.Set(); });
  447. Task.Run(() =>
  448. {
  449. s.OnNext(1);
  450. next.WaitOne();
  451. s.OnError(ex);
  452. });
  453. e.WaitOne();
  454. Assert.Same(ex, err);
  455. }
  456. #if !NO_THREAD
  457. [Fact]
  458. public void Delay_TimeSpan_Real_Error3()
  459. {
  460. Delay_TimeSpan_Real_Error3_Impl(ThreadPoolScheduler.Instance.DisableOptimizations());
  461. }
  462. [Fact]
  463. public void Delay_TimeSpan_Real_Error3_Stopwatch()
  464. {
  465. Delay_TimeSpan_Real_Error3_Impl(ThreadPoolScheduler.Instance);
  466. }
  467. #endif
  468. private void Delay_TimeSpan_Real_Error3_Impl(IScheduler scheduler)
  469. {
  470. var ex = new Exception();
  471. var s = new Subject<int>();
  472. var res = s.Delay(TimeSpan.FromMilliseconds(10), scheduler);
  473. var next = new ManualResetEvent(false);
  474. var ack = new ManualResetEvent(false);
  475. var e = new ManualResetEvent(false);
  476. var err = default(Exception);
  477. res.Subscribe(_ => { next.Set(); ack.WaitOne(); }, ex_ => { err = ex_; e.Set(); });
  478. Task.Run(() =>
  479. {
  480. s.OnNext(1);
  481. next.WaitOne();
  482. s.OnError(ex);
  483. ack.Set();
  484. });
  485. e.WaitOne();
  486. Assert.Same(ex, err);
  487. }
  488. [Fact]
  489. public void Delay_TimeSpan_Positive()
  490. {
  491. var scheduler = new TestScheduler();
  492. var msgs = new[] {
  493. OnNext(150, 1),
  494. OnNext(250, 2),
  495. OnNext(350, 3),
  496. OnNext(450, 4),
  497. OnCompleted<int>(550)
  498. };
  499. var xs = scheduler.CreateHotObservable(msgs);
  500. const ushort delay = 42;
  501. var res = scheduler.Start(() =>
  502. xs.Delay(TimeSpan.FromTicks(delay), scheduler)
  503. );
  504. var expected = from n in msgs
  505. where n.Time > Subscribed
  506. select new Recorded<Notification<int>>((ushort)(n.Time + delay), n.Value);
  507. res.Messages.AssertEqual(expected);
  508. }
  509. [Fact]
  510. public void Delay_Empty()
  511. {
  512. var scheduler = new TestScheduler();
  513. var xs = scheduler.CreateHotObservable(
  514. OnNext(150, 1),
  515. OnCompleted<int>(550)
  516. );
  517. var res = scheduler.Start(() =>
  518. xs.Delay(TimeSpan.FromTicks(10), scheduler)
  519. );
  520. res.Messages.AssertEqual(
  521. OnCompleted<int>(560)
  522. );
  523. xs.Subscriptions.AssertEqual(
  524. Subscribe(200, 550)
  525. );
  526. }
  527. [Fact]
  528. public void Delay_Error()
  529. {
  530. var scheduler = new TestScheduler();
  531. var ex = new Exception();
  532. var xs = scheduler.CreateHotObservable(
  533. OnNext(150, 1),
  534. OnError<int>(550, ex)
  535. );
  536. var res = scheduler.Start(() =>
  537. xs.Delay(TimeSpan.FromTicks(10), scheduler)
  538. );
  539. res.Messages.AssertEqual(
  540. OnError<int>(550, ex)
  541. );
  542. xs.Subscriptions.AssertEqual(
  543. Subscribe(200, 550)
  544. );
  545. }
  546. [Fact]
  547. public void Delay_Never()
  548. {
  549. var scheduler = new TestScheduler();
  550. var xs = scheduler.CreateHotObservable(
  551. OnNext(150, 1)
  552. );
  553. var res = scheduler.Start(() =>
  554. xs.Delay(TimeSpan.FromTicks(10), scheduler)
  555. );
  556. res.Messages.AssertEqual(
  557. );
  558. xs.Subscriptions.AssertEqual(
  559. Subscribe(200, 1000)
  560. );
  561. }
  562. [Fact]
  563. public void Delay_TimeSpan_DefaultScheduler()
  564. {
  565. Assert.True(Observable.Return(1).Delay(TimeSpan.FromMilliseconds(1)).ToEnumerable().SequenceEqual(new[] { 1 }));
  566. }
  567. [Fact]
  568. public void Delay_DateTimeOffset_DefaultScheduler()
  569. {
  570. Assert.True(Observable.Return(1).Delay(DateTimeOffset.UtcNow + TimeSpan.FromMilliseconds(1)).ToEnumerable().SequenceEqual(new[] { 1 }));
  571. }
  572. [Fact]
  573. public void Delay_CrossingMessages()
  574. {
  575. var lst = new List<int>();
  576. var evt = new ManualResetEvent(false);
  577. var s = new Subject<int>();
  578. s.Delay(TimeSpan.FromSeconds(0.01)).Subscribe(x =>
  579. {
  580. lst.Add(x);
  581. if (x < 9)
  582. {
  583. s.OnNext(x + 1);
  584. }
  585. else
  586. {
  587. s.OnCompleted();
  588. }
  589. }, () =>
  590. {
  591. evt.Set();
  592. });
  593. s.OnNext(0);
  594. evt.WaitOne();
  595. Assert.True(Enumerable.Range(0, 10).SequenceEqual(lst));
  596. }
  597. [Fact]
  598. public void Delay_Duration_ArgumentChecking()
  599. {
  600. var someObservable = DummyObservable<int>.Instance;
  601. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(default(IObservable<int>), x => someObservable));
  602. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(someObservable, default(Func<int, IObservable<int>>)));
  603. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(default(IObservable<int>), someObservable, x => someObservable));
  604. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(someObservable, default, x => someObservable));
  605. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Delay(someObservable, someObservable, default));
  606. }
  607. [Fact]
  608. public void Delay_Duration_Simple1()
  609. {
  610. var scheduler = new TestScheduler();
  611. var xs = scheduler.CreateHotObservable(
  612. OnNext(150, 1),
  613. OnNext(210, 10),
  614. OnNext(220, 30),
  615. OnNext(230, 50),
  616. OnNext(240, 35),
  617. OnNext(250, 20),
  618. OnCompleted<int>(260)
  619. );
  620. var res = scheduler.Start(() =>
  621. xs.Delay(x => scheduler.CreateColdObservable(new[] { OnNext(x, "!") }))
  622. );
  623. res.Messages.AssertEqual(
  624. OnNext(210 + 10, 10),
  625. OnNext(220 + 30, 30),
  626. OnNext(250 + 20, 20),
  627. OnNext(240 + 35, 35),
  628. OnNext(230 + 50, 50),
  629. OnCompleted<int>(280)
  630. );
  631. xs.Subscriptions.AssertEqual(
  632. Subscribe(200, 260)
  633. );
  634. }
  635. [Fact]
  636. public void Delay_Duration_Simple2()
  637. {
  638. var scheduler = new TestScheduler();
  639. var xs = scheduler.CreateHotObservable(
  640. OnNext(150, 1),
  641. OnNext(210, 2),
  642. OnNext(220, 3),
  643. OnNext(230, 4),
  644. OnNext(240, 5),
  645. OnNext(250, 6),
  646. OnCompleted<int>(300)
  647. );
  648. var ys = scheduler.CreateColdObservable(
  649. OnNext(10, "!")
  650. );
  651. var res = scheduler.Start(() =>
  652. xs.Delay(_ => ys)
  653. );
  654. res.Messages.AssertEqual(
  655. OnNext(210 + 10, 2),
  656. OnNext(220 + 10, 3),
  657. OnNext(230 + 10, 4),
  658. OnNext(240 + 10, 5),
  659. OnNext(250 + 10, 6),
  660. OnCompleted<int>(300)
  661. );
  662. xs.Subscriptions.AssertEqual(
  663. Subscribe(200, 300)
  664. );
  665. ys.Subscriptions.AssertEqual(
  666. Subscribe(210, 220),
  667. Subscribe(220, 230),
  668. Subscribe(230, 240),
  669. Subscribe(240, 250),
  670. Subscribe(250, 260)
  671. );
  672. }
  673. [Fact]
  674. public void Delay_Duration_Simple3()
  675. {
  676. var scheduler = new TestScheduler();
  677. var xs = scheduler.CreateHotObservable(
  678. OnNext(150, 1),
  679. OnNext(210, 2),
  680. OnNext(220, 3),
  681. OnNext(230, 4),
  682. OnNext(240, 5),
  683. OnNext(250, 6),
  684. OnCompleted<int>(300)
  685. );
  686. var ys = scheduler.CreateColdObservable(
  687. OnNext(100, "!")
  688. );
  689. var res = scheduler.Start(() =>
  690. xs.Delay(_ => ys)
  691. );
  692. res.Messages.AssertEqual(
  693. OnNext(210 + 100, 2),
  694. OnNext(220 + 100, 3),
  695. OnNext(230 + 100, 4),
  696. OnNext(240 + 100, 5),
  697. OnNext(250 + 100, 6),
  698. OnCompleted<int>(350)
  699. );
  700. xs.Subscriptions.AssertEqual(
  701. Subscribe(200, 300)
  702. );
  703. ys.Subscriptions.AssertEqual(
  704. Subscribe(210, 310),
  705. Subscribe(220, 320),
  706. Subscribe(230, 330),
  707. Subscribe(240, 340),
  708. Subscribe(250, 350)
  709. );
  710. }
  711. [Fact]
  712. public void Delay_Duration_Simple4_InnerEmpty()
  713. {
  714. var scheduler = new TestScheduler();
  715. var xs = scheduler.CreateHotObservable(
  716. OnNext(150, 1),
  717. OnNext(210, 2),
  718. OnNext(220, 3),
  719. OnNext(230, 4),
  720. OnNext(240, 5),
  721. OnNext(250, 6),
  722. OnCompleted<int>(300)
  723. );
  724. var ys = scheduler.CreateColdObservable(
  725. OnCompleted<int>(100)
  726. );
  727. var res = scheduler.Start(() =>
  728. xs.Delay(_ => ys)
  729. );
  730. res.Messages.AssertEqual(
  731. OnNext(210 + 100, 2),
  732. OnNext(220 + 100, 3),
  733. OnNext(230 + 100, 4),
  734. OnNext(240 + 100, 5),
  735. OnNext(250 + 100, 6),
  736. OnCompleted<int>(350)
  737. );
  738. xs.Subscriptions.AssertEqual(
  739. Subscribe(200, 300)
  740. );
  741. ys.Subscriptions.AssertEqual(
  742. Subscribe(210, 310),
  743. Subscribe(220, 320),
  744. Subscribe(230, 330),
  745. Subscribe(240, 340),
  746. Subscribe(250, 350)
  747. );
  748. }
  749. [Fact]
  750. public void Delay_Duration_Dispose1()
  751. {
  752. var scheduler = new TestScheduler();
  753. var xs = scheduler.CreateHotObservable(
  754. OnNext(150, 1),
  755. OnNext(210, 2),
  756. OnNext(220, 3),
  757. OnNext(230, 4),
  758. OnNext(240, 5),
  759. OnNext(250, 6),
  760. OnCompleted<int>(300)
  761. );
  762. var ys = scheduler.CreateColdObservable(
  763. OnNext(200, "!")
  764. );
  765. var res = scheduler.Start(() =>
  766. xs.Delay(_ => ys),
  767. 425
  768. );
  769. res.Messages.AssertEqual(
  770. OnNext(210 + 200, 2),
  771. OnNext(220 + 200, 3)
  772. );
  773. xs.Subscriptions.AssertEqual(
  774. Subscribe(200, 300)
  775. );
  776. ys.Subscriptions.AssertEqual(
  777. Subscribe(210, 410),
  778. Subscribe(220, 420),
  779. Subscribe(230, 425),
  780. Subscribe(240, 425),
  781. Subscribe(250, 425)
  782. );
  783. }
  784. [Fact]
  785. public void Delay_Duration_Dispose2()
  786. {
  787. var scheduler = new TestScheduler();
  788. var xs = scheduler.CreateHotObservable(
  789. OnNext(150, 1),
  790. OnNext(210, 2),
  791. OnNext(400, 3),
  792. OnCompleted<int>(500)
  793. );
  794. var ys = scheduler.CreateColdObservable(
  795. OnNext(50, "!")
  796. );
  797. var res = scheduler.Start(() =>
  798. xs.Delay(_ => ys),
  799. 300
  800. );
  801. res.Messages.AssertEqual(
  802. OnNext(210 + 50, 2)
  803. );
  804. xs.Subscriptions.AssertEqual(
  805. Subscribe(200, 300)
  806. );
  807. ys.Subscriptions.AssertEqual(
  808. Subscribe(210, 260)
  809. );
  810. }
  811. [Fact]
  812. public void Delay_Duration_OuterError1()
  813. {
  814. var scheduler = new TestScheduler();
  815. var ex = new Exception();
  816. var xs = scheduler.CreateHotObservable(
  817. OnNext(150, 1),
  818. OnNext(210, 2),
  819. OnNext(220, 3),
  820. OnNext(230, 4),
  821. OnNext(240, 5),
  822. OnNext(250, 6),
  823. OnError<int>(300, ex)
  824. );
  825. var ys = scheduler.CreateColdObservable(
  826. OnNext(100, "!")
  827. );
  828. var res = scheduler.Start(() =>
  829. xs.Delay(_ => ys)
  830. );
  831. res.Messages.AssertEqual(
  832. OnError<int>(300, ex)
  833. );
  834. xs.Subscriptions.AssertEqual(
  835. Subscribe(200, 300)
  836. );
  837. ys.Subscriptions.AssertEqual(
  838. Subscribe(210, 300),
  839. Subscribe(220, 300),
  840. Subscribe(230, 300),
  841. Subscribe(240, 300),
  842. Subscribe(250, 300)
  843. );
  844. }
  845. [Fact]
  846. public void Delay_Duration_OuterError2()
  847. {
  848. var scheduler = new TestScheduler();
  849. var ex = new Exception();
  850. var xs = scheduler.CreateHotObservable(
  851. OnNext(150, 1),
  852. OnNext(210, 2),
  853. OnNext(220, 3),
  854. OnNext(230, 4),
  855. OnNext(240, 5),
  856. OnNext(250, 6),
  857. OnError<int>(300, ex)
  858. );
  859. var ys = scheduler.CreateColdObservable(
  860. OnNext(10, "!")
  861. );
  862. var res = scheduler.Start(() =>
  863. xs.Delay(_ => ys)
  864. );
  865. res.Messages.AssertEqual(
  866. OnNext(210 + 10, 2),
  867. OnNext(220 + 10, 3),
  868. OnNext(230 + 10, 4),
  869. OnNext(240 + 10, 5),
  870. OnNext(250 + 10, 6),
  871. OnError<int>(300, ex)
  872. );
  873. xs.Subscriptions.AssertEqual(
  874. Subscribe(200, 300)
  875. );
  876. ys.Subscriptions.AssertEqual(
  877. Subscribe(210, 220),
  878. Subscribe(220, 230),
  879. Subscribe(230, 240),
  880. Subscribe(240, 250),
  881. Subscribe(250, 260)
  882. );
  883. }
  884. [Fact]
  885. public void Delay_Duration_InnerError1()
  886. {
  887. var scheduler = new TestScheduler();
  888. var ex = new Exception();
  889. var xs = scheduler.CreateHotObservable(
  890. OnNext(150, 1),
  891. OnNext(210, 2),
  892. OnNext(220, 3),
  893. OnNext(230, 4),
  894. OnNext(240, 5),
  895. OnNext(250, 6),
  896. OnCompleted<int>(300)
  897. );
  898. var ys = scheduler.CreateColdObservable(
  899. OnNext(30, "!")
  900. );
  901. var zs = scheduler.CreateColdObservable(
  902. OnError<string>(25, ex)
  903. );
  904. var res = scheduler.Start(() =>
  905. xs.Delay(x => x != 5 ? ys : zs)
  906. );
  907. res.Messages.AssertEqual(
  908. OnNext(210 + 30, 2),
  909. OnNext(220 + 30, 3),
  910. OnNext(230 + 30, 4),
  911. OnError<int>(240 + 25, ex)
  912. );
  913. xs.Subscriptions.AssertEqual(
  914. Subscribe(200, 265)
  915. );
  916. ys.Subscriptions.AssertEqual(
  917. Subscribe(210, 240),
  918. Subscribe(220, 250),
  919. Subscribe(230, 260),
  920. Subscribe(250, 265)
  921. );
  922. }
  923. [Fact]
  924. public void Delay_Duration_InnerError2()
  925. {
  926. var scheduler = new TestScheduler();
  927. var ex = new Exception();
  928. var xs = scheduler.CreateHotObservable(
  929. OnNext(150, 1),
  930. OnNext(210, 2),
  931. OnNext(220, 3),
  932. OnNext(230, 4),
  933. OnNext(240, 5),
  934. OnNext(250, 6),
  935. OnCompleted<int>(300)
  936. );
  937. var ys = scheduler.CreateColdObservable(
  938. OnError<string>(100, ex)
  939. );
  940. var res = scheduler.Start(() =>
  941. xs.Delay(_ => ys)
  942. );
  943. res.Messages.AssertEqual(
  944. OnError<int>(210 + 100, ex)
  945. );
  946. xs.Subscriptions.AssertEqual(
  947. Subscribe(200, 300)
  948. );
  949. ys.Subscriptions.AssertEqual(
  950. Subscribe(210, 310),
  951. Subscribe(220, 310),
  952. Subscribe(230, 310),
  953. Subscribe(240, 310),
  954. Subscribe(250, 310)
  955. );
  956. }
  957. [Fact]
  958. public void Delay_Duration_SelectorThrows1()
  959. {
  960. var scheduler = new TestScheduler();
  961. var ex = new Exception();
  962. var xs = scheduler.CreateHotObservable(
  963. OnNext(150, 1),
  964. OnNext(250, 2),
  965. OnNext(300, 3),
  966. OnNext(350, 4),
  967. OnNext(400, 5),
  968. OnNext(450, 6),
  969. OnCompleted<int>(500)
  970. );
  971. var ys = scheduler.CreateColdObservable(
  972. OnNext(80, "")
  973. );
  974. var res = scheduler.Start(() =>
  975. xs.Delay(x =>
  976. {
  977. if (x == 4)
  978. {
  979. throw ex;
  980. }
  981. return ys;
  982. })
  983. );
  984. res.Messages.AssertEqual(
  985. OnNext(330, 2),
  986. OnError<int>(350, ex)
  987. );
  988. xs.Subscriptions.AssertEqual(
  989. Subscribe(200, 350)
  990. );
  991. ys.Subscriptions.AssertEqual(
  992. Subscribe(250, 330),
  993. Subscribe(300, 350)
  994. );
  995. }
  996. [Fact]
  997. public void Delay_Duration_Simple()
  998. {
  999. var scheduler = new TestScheduler();
  1000. var xs = scheduler.CreateHotObservable(
  1001. OnNext(150, 1),
  1002. OnNext(250, 2),
  1003. OnNext(350, 3),
  1004. OnNext(450, 4),
  1005. OnCompleted<int>(550)
  1006. );
  1007. var res = scheduler.Start(() =>
  1008. xs.Delay(x =>
  1009. scheduler.CreateColdObservable(
  1010. OnNext(x * 10, "Ignore"),
  1011. OnNext(x * 10 + 5, "Aargh!")
  1012. )
  1013. )
  1014. );
  1015. res.Messages.AssertEqual(
  1016. OnNext(250 + 2 * 10, 2),
  1017. OnNext(350 + 3 * 10, 3),
  1018. OnNext(450 + 4 * 10, 4),
  1019. OnCompleted<int>(550)
  1020. );
  1021. xs.Subscriptions.AssertEqual(
  1022. Subscribe(200, 550)
  1023. );
  1024. }
  1025. [Fact]
  1026. public void Delay_Duration_DeferOnCompleted()
  1027. {
  1028. var scheduler = new TestScheduler();
  1029. var xs = scheduler.CreateHotObservable(
  1030. OnNext(150, 1),
  1031. OnNext(250, 2),
  1032. OnNext(350, 3),
  1033. OnNext(450, 4),
  1034. OnCompleted<int>(451)
  1035. );
  1036. var res = scheduler.Start(() =>
  1037. xs.Delay(x =>
  1038. scheduler.CreateColdObservable(
  1039. OnNext(x * 10, "Ignore"),
  1040. OnNext(x * 10 + 5, "Aargh!")
  1041. )
  1042. )
  1043. );
  1044. res.Messages.AssertEqual(
  1045. OnNext(250 + 2 * 10, 2),
  1046. OnNext(350 + 3 * 10, 3),
  1047. OnNext(450 + 4 * 10, 4),
  1048. OnCompleted<int>(450 + 4 * 10)
  1049. );
  1050. xs.Subscriptions.AssertEqual(
  1051. Subscribe(200, 451)
  1052. );
  1053. }
  1054. [Fact]
  1055. public void Delay_Duration_InnerError()
  1056. {
  1057. var scheduler = new TestScheduler();
  1058. var xs = scheduler.CreateHotObservable(
  1059. OnNext(150, 1),
  1060. OnNext(250, 2),
  1061. OnNext(350, 3),
  1062. OnNext(450, 4),
  1063. OnCompleted<int>(451)
  1064. );
  1065. var ex = new Exception();
  1066. var res = scheduler.Start(() =>
  1067. xs.Delay(x =>
  1068. x < 4 ? scheduler.CreateColdObservable(
  1069. OnNext(x * 10, "Ignore"),
  1070. OnNext(x * 10 + 5, "Aargh!")
  1071. )
  1072. : scheduler.CreateColdObservable(
  1073. OnError<string>(x * 10, ex)
  1074. )
  1075. )
  1076. );
  1077. res.Messages.AssertEqual(
  1078. OnNext(250 + 2 * 10, 2),
  1079. OnNext(350 + 3 * 10, 3),
  1080. OnError<int>(450 + 4 * 10, ex)
  1081. );
  1082. xs.Subscriptions.AssertEqual(
  1083. Subscribe(200, 451)
  1084. );
  1085. }
  1086. [Fact]
  1087. public void Delay_Duration_OuterError()
  1088. {
  1089. var scheduler = new TestScheduler();
  1090. var ex = new Exception();
  1091. var xs = scheduler.CreateHotObservable(
  1092. OnNext(150, 1),
  1093. OnNext(250, 2),
  1094. OnNext(350, 3),
  1095. OnNext(450, 4),
  1096. OnError<int>(460, ex)
  1097. );
  1098. var res = scheduler.Start(() =>
  1099. xs.Delay(x =>
  1100. scheduler.CreateColdObservable(
  1101. OnNext(x * 10, "Ignore"),
  1102. OnNext(x * 10 + 5, "Aargh!")
  1103. )
  1104. )
  1105. );
  1106. res.Messages.AssertEqual(
  1107. OnNext(250 + 2 * 10, 2),
  1108. OnNext(350 + 3 * 10, 3),
  1109. OnError<int>(460, ex)
  1110. );
  1111. xs.Subscriptions.AssertEqual(
  1112. Subscribe(200, 460)
  1113. );
  1114. }
  1115. [Fact]
  1116. public void Delay_Duration_SelectorThrows2()
  1117. {
  1118. var scheduler = new TestScheduler();
  1119. var xs = scheduler.CreateHotObservable(
  1120. OnNext(150, 1),
  1121. OnNext(250, 2),
  1122. OnNext(350, 3),
  1123. OnNext(450, 4),
  1124. OnCompleted<int>(550)
  1125. );
  1126. var ex = new Exception();
  1127. var res = scheduler.Start(() =>
  1128. xs.Delay(x =>
  1129. {
  1130. if (x < 4)
  1131. {
  1132. return scheduler.CreateColdObservable(
  1133. OnNext(x * 10, "Ignore"),
  1134. OnNext(x * 10 + 5, "Aargh!")
  1135. );
  1136. }
  1137. else
  1138. {
  1139. throw ex;
  1140. }
  1141. })
  1142. );
  1143. res.Messages.AssertEqual(
  1144. OnNext(250 + 2 * 10, 2),
  1145. OnNext(350 + 3 * 10, 3),
  1146. OnError<int>(450, ex)
  1147. );
  1148. xs.Subscriptions.AssertEqual(
  1149. Subscribe(200, 450)
  1150. );
  1151. }
  1152. [Fact]
  1153. public void Delay_Duration_InnerDone()
  1154. {
  1155. var scheduler = new TestScheduler();
  1156. var xs = scheduler.CreateHotObservable(
  1157. OnNext(150, 1),
  1158. OnNext(250, 2),
  1159. OnNext(350, 3),
  1160. OnNext(450, 4),
  1161. OnCompleted<int>(550)
  1162. );
  1163. var ex = new Exception();
  1164. var res = scheduler.Start(() =>
  1165. xs.Delay(x =>
  1166. scheduler.CreateColdObservable(
  1167. OnCompleted<string>(x * 10)
  1168. )
  1169. )
  1170. );
  1171. res.Messages.AssertEqual(
  1172. OnNext(250 + 2 * 10, 2),
  1173. OnNext(350 + 3 * 10, 3),
  1174. OnNext(450 + 4 * 10, 4),
  1175. OnCompleted<int>(550)
  1176. );
  1177. xs.Subscriptions.AssertEqual(
  1178. Subscribe(200, 550)
  1179. );
  1180. }
  1181. [Fact]
  1182. public void Delay_Duration_InnerSubscriptionTimes()
  1183. {
  1184. var scheduler = new TestScheduler();
  1185. var xs = scheduler.CreateHotObservable(
  1186. OnNext(150, -1),
  1187. OnNext(250, 0),
  1188. OnNext(350, 1),
  1189. OnNext(450, 2),
  1190. OnCompleted<int>(550)
  1191. );
  1192. var ys = new[] {
  1193. scheduler.CreateColdObservable(
  1194. OnNext(20, 42),
  1195. OnNext(25, 99)
  1196. ),
  1197. scheduler.CreateColdObservable(
  1198. OnNext(10, 43),
  1199. OnNext(15, 99)
  1200. ),
  1201. scheduler.CreateColdObservable(
  1202. OnNext(30, 44),
  1203. OnNext(35, 99)
  1204. ),
  1205. };
  1206. var res = scheduler.Start(() =>
  1207. xs.Delay(x => ys[x])
  1208. );
  1209. res.Messages.AssertEqual(
  1210. OnNext(250 + 20, 0),
  1211. OnNext(350 + 10, 1),
  1212. OnNext(450 + 30, 2),
  1213. OnCompleted<int>(550)
  1214. );
  1215. xs.Subscriptions.AssertEqual(
  1216. Subscribe(200, 550)
  1217. );
  1218. ys[0].Subscriptions.AssertEqual(Subscribe(250, 250 + 20));
  1219. ys[1].Subscriptions.AssertEqual(Subscribe(350, 350 + 10));
  1220. ys[2].Subscriptions.AssertEqual(Subscribe(450, 450 + 30));
  1221. }
  1222. [Fact]
  1223. public void Delay_DurationAndSubscription_Simple1()
  1224. {
  1225. var scheduler = new TestScheduler();
  1226. var xs = scheduler.CreateHotObservable(
  1227. OnNext(150, 1),
  1228. OnNext(210, 10),
  1229. OnNext(220, 30),
  1230. OnNext(230, 50),
  1231. OnNext(240, 35),
  1232. OnNext(250, 20),
  1233. OnCompleted<int>(260)
  1234. );
  1235. var ys = scheduler.CreateColdObservable(
  1236. OnNext(10, "!")
  1237. );
  1238. var res = scheduler.Start(() =>
  1239. xs.Delay(ys, x => scheduler.CreateColdObservable(new[] { OnNext(x, "!") }))
  1240. );
  1241. res.Messages.AssertEqual(
  1242. OnNext(220 + 30, 30),
  1243. OnNext(250 + 20, 20),
  1244. OnNext(240 + 35, 35),
  1245. OnNext(230 + 50, 50),
  1246. OnCompleted<int>(280)
  1247. );
  1248. xs.Subscriptions.AssertEqual(
  1249. Subscribe(210, 260)
  1250. );
  1251. ys.Subscriptions.AssertEqual(
  1252. Subscribe(200, 210)
  1253. );
  1254. }
  1255. [Fact]
  1256. public void Delay_DurationAndSubscription_Simple2()
  1257. {
  1258. var scheduler = new TestScheduler();
  1259. var xs = scheduler.CreateHotObservable(
  1260. OnNext(150, 1),
  1261. OnNext(210, 10),
  1262. OnNext(220, 30),
  1263. OnNext(230, 50),
  1264. OnNext(240, 35),
  1265. OnNext(250, 20),
  1266. OnCompleted<int>(260)
  1267. );
  1268. var ys = scheduler.CreateColdObservable(
  1269. OnCompleted<string>(10)
  1270. );
  1271. var res = scheduler.Start(() =>
  1272. xs.Delay(ys, x => scheduler.CreateColdObservable(new[] { OnNext(x, "!") }))
  1273. );
  1274. res.Messages.AssertEqual(
  1275. OnNext(220 + 30, 30),
  1276. OnNext(250 + 20, 20),
  1277. OnNext(240 + 35, 35),
  1278. OnNext(230 + 50, 50),
  1279. OnCompleted<int>(280)
  1280. );
  1281. xs.Subscriptions.AssertEqual(
  1282. Subscribe(210, 260)
  1283. );
  1284. ys.Subscriptions.AssertEqual(
  1285. Subscribe(200, 210)
  1286. );
  1287. }
  1288. [Fact]
  1289. public void Delay_DurationAndSubscription_Dispose1()
  1290. {
  1291. var scheduler = new TestScheduler();
  1292. var xs = scheduler.CreateHotObservable(
  1293. OnNext(150, 1),
  1294. OnNext(210, 10),
  1295. OnNext(220, 30),
  1296. OnNext(230, 50),
  1297. OnNext(240, 35),
  1298. OnNext(250, 20),
  1299. OnCompleted<int>(260)
  1300. );
  1301. var ys = scheduler.CreateColdObservable(
  1302. OnNext(10, "!")
  1303. );
  1304. var res = scheduler.Start(() =>
  1305. xs.Delay(ys, x => scheduler.CreateColdObservable(new[] { OnNext(x, "!") })),
  1306. 255
  1307. );
  1308. res.Messages.AssertEqual(
  1309. OnNext(220 + 30, 30)
  1310. );
  1311. xs.Subscriptions.AssertEqual(
  1312. Subscribe(210, 255)
  1313. );
  1314. ys.Subscriptions.AssertEqual(
  1315. Subscribe(200, 210)
  1316. );
  1317. }
  1318. [Fact]
  1319. public void Delay_DurationAndSubscription_Dispose2()
  1320. {
  1321. var scheduler = new TestScheduler();
  1322. var xs = scheduler.CreateHotObservable(
  1323. OnNext(150, 1),
  1324. OnNext(210, 10),
  1325. OnNext(220, 30),
  1326. OnNext(230, 50),
  1327. OnNext(240, 35),
  1328. OnNext(250, 20),
  1329. OnCompleted<int>(260)
  1330. );
  1331. var ys = scheduler.CreateColdObservable(
  1332. OnNext(100, "!")
  1333. );
  1334. var res = scheduler.Start(() =>
  1335. xs.Delay(ys, x => scheduler.CreateColdObservable(new[] { OnNext(x, "!") })),
  1336. 255
  1337. );
  1338. res.Messages.AssertEqual(
  1339. );
  1340. xs.Subscriptions.AssertEqual(
  1341. );
  1342. ys.Subscriptions.AssertEqual(
  1343. Subscribe(200, 255)
  1344. );
  1345. }
  1346. [Fact]
  1347. public void Delay_DurationAndSubscription_Error()
  1348. {
  1349. var scheduler = new TestScheduler();
  1350. var ex = new Exception();
  1351. var xs = scheduler.CreateHotObservable(
  1352. OnNext(150, 1),
  1353. OnNext(210, 10),
  1354. OnNext(220, 30),
  1355. OnNext(230, 50),
  1356. OnNext(240, 35),
  1357. OnNext(250, 20),
  1358. OnCompleted<int>(260)
  1359. );
  1360. var ys = scheduler.CreateColdObservable(
  1361. OnError<string>(10, ex)
  1362. );
  1363. var res = scheduler.Start(() =>
  1364. xs.Delay(ys, x => scheduler.CreateColdObservable(new[] { OnNext(x, "!") }))
  1365. );
  1366. res.Messages.AssertEqual(
  1367. OnError<int>(200 + 10, ex)
  1368. );
  1369. xs.Subscriptions.AssertEqual(
  1370. );
  1371. ys.Subscriptions.AssertEqual(
  1372. Subscribe(200, 210)
  1373. );
  1374. }
  1375. [Fact]
  1376. public void Delay_ErrorHandling1()
  1377. {
  1378. //
  1379. // Checks for race condition between OnNext and OnError where the latter has a chance to
  1380. // send out the OnError message before the former gets a chance to run in the delayed
  1381. // queue. In that case, the OnNext message should not come out.
  1382. //
  1383. // See DrainQueue's first _hasFailed check.
  1384. //
  1385. var xs = Observable.Create<int>(observer =>
  1386. {
  1387. observer.OnNext(42);
  1388. observer.OnError(new Exception());
  1389. return () => { };
  1390. });
  1391. var s = new ImpulseScheduler();
  1392. var called = false;
  1393. var failed = new ManualResetEvent(false);
  1394. xs.Delay(TimeSpan.FromDays(1), s).Subscribe(_ => { called = true; }, ex => { failed.Set(); });
  1395. failed.WaitOne();
  1396. s.Event.Set();
  1397. s.Done.WaitOne();
  1398. Assert.False(called);
  1399. }
  1400. private class ImpulseScheduler : IScheduler
  1401. {
  1402. public DateTimeOffset Now
  1403. {
  1404. get { return DateTimeOffset.UtcNow; }
  1405. }
  1406. public IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
  1407. {
  1408. throw new NotImplementedException();
  1409. }
  1410. private ManualResetEvent _event = new ManualResetEvent(false);
  1411. private ManualResetEvent _done = new ManualResetEvent(false);
  1412. public ManualResetEvent Event { get { return _event; } }
  1413. public ManualResetEvent Done { get { return _done; } }
  1414. public IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  1415. {
  1416. Scheduler.Default.Schedule(() =>
  1417. {
  1418. _event.WaitOne();
  1419. action(this, state);
  1420. _done.Set();
  1421. });
  1422. return Disposable.Empty;
  1423. }
  1424. public IDisposable Schedule<TState>(TState state, DateTimeOffset dueTime, Func<IScheduler, TState, IDisposable> action)
  1425. {
  1426. throw new NotImplementedException();
  1427. }
  1428. }
  1429. [Fact]
  1430. public void Delay_LongRunning_CancelEarly()
  1431. {
  1432. var xs = Observable.Create<int>(observer =>
  1433. {
  1434. return Scheduler.Default.Schedule(TimeSpan.FromHours(1), () =>
  1435. {
  1436. observer.OnNext(42);
  1437. });
  1438. });
  1439. var s = new ManualResetEvent(false);
  1440. var e = new ManualResetEvent(false);
  1441. var ys = xs.Delay(TimeSpan.Zero, new MyLongRunning1(s, e));
  1442. var d = ys.Subscribe(_ => { });
  1443. s.WaitOne();
  1444. d.Dispose();
  1445. e.WaitOne();
  1446. }
  1447. private class MyLongRunning1 : LocalScheduler, ISchedulerLongRunning
  1448. {
  1449. private ManualResetEvent _start;
  1450. private ManualResetEvent _stop;
  1451. public MyLongRunning1(ManualResetEvent start, ManualResetEvent stop)
  1452. {
  1453. _start = start;
  1454. _stop = stop;
  1455. }
  1456. public IDisposable ScheduleLongRunning<TState>(TState state, Action<TState, ICancelable> action)
  1457. {
  1458. var b = new BooleanDisposable();
  1459. Task.Run(() =>
  1460. {
  1461. _start.Set();
  1462. action(state, b);
  1463. _stop.Set();
  1464. });
  1465. return b;
  1466. }
  1467. public override IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  1468. {
  1469. throw new NotImplementedException();
  1470. }
  1471. }
  1472. [Fact]
  1473. public void Delay_LongRunning_CancelLate()
  1474. {
  1475. var xs = Observable.Return(42);
  1476. var s = new ManualResetEvent(false);
  1477. var e = new ManualResetEvent(false);
  1478. var ys = xs.Delay(TimeSpan.FromHours(1), new MyLongRunning2(s, e));
  1479. var d = ys.Subscribe(_ => { });
  1480. s.WaitOne();
  1481. d.Dispose();
  1482. e.WaitOne();
  1483. }
  1484. private class MyLongRunning2 : LocalScheduler, ISchedulerLongRunning
  1485. {
  1486. private ManualResetEvent _start;
  1487. private ManualResetEvent _stop;
  1488. public MyLongRunning2(ManualResetEvent start, ManualResetEvent stop)
  1489. {
  1490. _start = start;
  1491. _stop = stop;
  1492. }
  1493. public IDisposable ScheduleLongRunning<TState>(TState state, Action<TState, ICancelable> action)
  1494. {
  1495. var b = new BooleanDisposable();
  1496. Task.Run(() =>
  1497. {
  1498. action(state, b);
  1499. _stop.Set();
  1500. });
  1501. return b;
  1502. }
  1503. public override IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  1504. {
  1505. _start.Set();
  1506. return Disposable.Empty;
  1507. }
  1508. }
  1509. }
  1510. }