TakeUntilTest.cs 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861
  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.Concurrency;
  8. using System.Reactive.Linq;
  9. using System.Threading;
  10. using Microsoft.Reactive.Testing;
  11. using ReactiveTests.Dummies;
  12. using Xunit;
  13. namespace ReactiveTests.Tests
  14. {
  15. public class TakeUntilTest : ReactiveTest
  16. {
  17. #region + Observable +
  18. [Fact]
  19. public void TakeUntil_ArgumentChecking()
  20. {
  21. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil<int, int>(null, DummyObservable<int>.Instance));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil<int, int>(DummyObservable<int>.Instance, null));
  23. }
  24. [Fact]
  25. public void TakeUntil_Preempt_SomeData_Next()
  26. {
  27. var scheduler = new TestScheduler();
  28. var l = scheduler.CreateHotObservable(
  29. OnNext(150, 1),
  30. OnNext(210, 2),
  31. OnNext(220, 3),
  32. OnNext(230, 4),
  33. OnNext(240, 5),
  34. OnCompleted<int>(250)
  35. );
  36. var r = scheduler.CreateHotObservable(
  37. OnNext(150, 1),
  38. OnNext(225, 99),
  39. OnCompleted<int>(230)
  40. );
  41. var res = scheduler.Start(() =>
  42. l.TakeUntil(r)
  43. );
  44. res.Messages.AssertEqual(
  45. OnNext(210, 2),
  46. OnNext(220, 3),
  47. OnCompleted<int>(225)
  48. );
  49. l.Subscriptions.AssertEqual(
  50. Subscribe(200, 225)
  51. );
  52. r.Subscriptions.AssertEqual(
  53. Subscribe(200, 225)
  54. );
  55. }
  56. [Fact]
  57. public void TakeUntil_Preempt_SomeData_Error()
  58. {
  59. var scheduler = new TestScheduler();
  60. var ex = new Exception();
  61. var l = scheduler.CreateHotObservable(
  62. OnNext(150, 1),
  63. OnNext(210, 2),
  64. OnNext(220, 3),
  65. OnNext(230, 4),
  66. OnNext(240, 5),
  67. OnCompleted<int>(250)
  68. );
  69. var r = scheduler.CreateHotObservable(
  70. OnNext(150, 1),
  71. OnError<int>(225, ex)
  72. );
  73. var res = scheduler.Start(() =>
  74. l.TakeUntil(r)
  75. );
  76. res.Messages.AssertEqual(
  77. OnNext(210, 2),
  78. OnNext(220, 3),
  79. OnError<int>(225, ex)
  80. );
  81. l.Subscriptions.AssertEqual(
  82. Subscribe(200, 225)
  83. );
  84. r.Subscriptions.AssertEqual(
  85. Subscribe(200, 225)
  86. );
  87. }
  88. [Fact]
  89. public void TakeUntil_NoPreempt_SomeData_Empty()
  90. {
  91. var scheduler = new TestScheduler();
  92. var l = scheduler.CreateHotObservable(
  93. OnNext(150, 1),
  94. OnNext(210, 2),
  95. OnNext(220, 3),
  96. OnNext(230, 4),
  97. OnNext(240, 5),
  98. OnCompleted<int>(250)
  99. );
  100. var r = scheduler.CreateHotObservable(
  101. OnNext(150, 1),
  102. OnCompleted<int>(225)
  103. );
  104. var res = scheduler.Start(() =>
  105. l.TakeUntil(r)
  106. );
  107. res.Messages.AssertEqual(
  108. OnNext(210, 2),
  109. OnNext(220, 3),
  110. OnNext(230, 4),
  111. OnNext(240, 5),
  112. OnCompleted<int>(250)
  113. );
  114. l.Subscriptions.AssertEqual(
  115. Subscribe(200, 250)
  116. );
  117. r.Subscriptions.AssertEqual(
  118. Subscribe(200, 225)
  119. );
  120. }
  121. [Fact]
  122. public void TakeUntil_NoPreempt_SomeData_Never()
  123. {
  124. var scheduler = new TestScheduler();
  125. var l = scheduler.CreateHotObservable(
  126. OnNext(150, 1),
  127. OnNext(210, 2),
  128. OnNext(220, 3),
  129. OnNext(230, 4),
  130. OnNext(240, 5),
  131. OnCompleted<int>(250)
  132. );
  133. var r = scheduler.CreateHotObservable(
  134. OnNext(150, 1)
  135. );
  136. var res = scheduler.Start(() =>
  137. l.TakeUntil(r)
  138. );
  139. res.Messages.AssertEqual(
  140. OnNext(210, 2),
  141. OnNext(220, 3),
  142. OnNext(230, 4),
  143. OnNext(240, 5),
  144. OnCompleted<int>(250)
  145. );
  146. l.Subscriptions.AssertEqual(
  147. Subscribe(200, 250)
  148. );
  149. r.Subscriptions.AssertEqual(
  150. Subscribe(200, 250)
  151. );
  152. }
  153. [Fact]
  154. public void TakeUntil_Preempt_Never_Next()
  155. {
  156. var scheduler = new TestScheduler();
  157. var l = scheduler.CreateHotObservable(
  158. OnNext(150, 1)
  159. );
  160. var r = scheduler.CreateHotObservable(
  161. OnNext(150, 1),
  162. OnNext(225, 2), //!
  163. OnCompleted<int>(250)
  164. );
  165. var res = scheduler.Start(() =>
  166. l.TakeUntil(r)
  167. );
  168. res.Messages.AssertEqual(
  169. OnCompleted<int>(225)
  170. );
  171. l.Subscriptions.AssertEqual(
  172. Subscribe(200, 225)
  173. );
  174. r.Subscriptions.AssertEqual(
  175. Subscribe(200, 225)
  176. );
  177. }
  178. [Fact]
  179. public void TakeUntil_Preempt_Never_Error()
  180. {
  181. var scheduler = new TestScheduler();
  182. var ex = new Exception();
  183. var l = scheduler.CreateHotObservable(
  184. OnNext(150, 1)
  185. );
  186. var r = scheduler.CreateHotObservable(
  187. OnNext(150, 1),
  188. OnError<int>(225, ex)
  189. );
  190. var res = scheduler.Start(() =>
  191. l.TakeUntil(r)
  192. );
  193. res.Messages.AssertEqual(
  194. OnError<int>(225, ex)
  195. );
  196. l.Subscriptions.AssertEqual(
  197. Subscribe(200, 225)
  198. );
  199. r.Subscriptions.AssertEqual(
  200. Subscribe(200, 225)
  201. );
  202. }
  203. [Fact]
  204. public void TakeUntil_NoPreempt_Never_Empty()
  205. {
  206. var scheduler = new TestScheduler();
  207. var l = scheduler.CreateHotObservable(
  208. OnNext(150, 1)
  209. );
  210. var r = scheduler.CreateHotObservable(
  211. OnNext(150, 1),
  212. OnCompleted<int>(225)
  213. );
  214. var res = scheduler.Start(() =>
  215. l.TakeUntil(r)
  216. );
  217. res.Messages.AssertEqual(
  218. );
  219. l.Subscriptions.AssertEqual(
  220. Subscribe(200, 1000 /* can't dispose prematurely, could be in flight to dispatch OnError */)
  221. );
  222. r.Subscriptions.AssertEqual(
  223. Subscribe(200, 225)
  224. );
  225. }
  226. [Fact]
  227. public void TakeUntil_NoPreempt_Never_Never()
  228. {
  229. var scheduler = new TestScheduler();
  230. var l = scheduler.CreateHotObservable(
  231. OnNext(150, 1)
  232. );
  233. var r = scheduler.CreateHotObservable(
  234. OnNext(150, 1)
  235. );
  236. var res = scheduler.Start(() =>
  237. l.TakeUntil(r)
  238. );
  239. res.Messages.AssertEqual(
  240. );
  241. l.Subscriptions.AssertEqual(
  242. Subscribe(200, 1000)
  243. );
  244. r.Subscriptions.AssertEqual(
  245. Subscribe(200, 1000)
  246. );
  247. }
  248. [Fact]
  249. public void TakeUntil_Preempt_BeforeFirstProduced()
  250. {
  251. var scheduler = new TestScheduler();
  252. var l = scheduler.CreateHotObservable(
  253. OnNext(150, 1),
  254. OnNext(230, 2),
  255. OnCompleted<int>(240)
  256. );
  257. var r = scheduler.CreateHotObservable(
  258. OnNext(150, 1),
  259. OnNext(210, 2), //!
  260. OnCompleted<int>(220)
  261. );
  262. var res = scheduler.Start(() =>
  263. l.TakeUntil(r)
  264. );
  265. res.Messages.AssertEqual(
  266. OnCompleted<int>(210)
  267. );
  268. l.Subscriptions.AssertEqual(
  269. Subscribe(200, 210)
  270. );
  271. r.Subscriptions.AssertEqual(
  272. Subscribe(200, 210)
  273. );
  274. }
  275. [Fact]
  276. public void TakeUntil_Preempt_BeforeFirstProduced_RemainSilentAndProperDisposed()
  277. {
  278. var scheduler = new TestScheduler();
  279. var sourceNotDisposed = false;
  280. var l = scheduler.CreateHotObservable(
  281. OnNext(150, 1),
  282. OnError<int>(215, new Exception()), // should not come
  283. OnCompleted<int>(240)
  284. );
  285. var r = scheduler.CreateHotObservable(
  286. OnNext(150, 1),
  287. OnNext(210, 2), //!
  288. OnCompleted<int>(220)
  289. );
  290. var res = scheduler.Start(() =>
  291. l.Do(_ => sourceNotDisposed = true).TakeUntil(r)
  292. );
  293. res.Messages.AssertEqual(
  294. OnCompleted<int>(210)
  295. );
  296. Assert.False(sourceNotDisposed);
  297. }
  298. [Fact]
  299. public void TakeUntil_NoPreempt_AfterLastProduced_ProperDisposedSignal()
  300. {
  301. var scheduler = new TestScheduler();
  302. var signalNotDisposed = false;
  303. var l = scheduler.CreateHotObservable(
  304. OnNext(150, 1),
  305. OnNext(230, 2),
  306. OnCompleted<int>(240)
  307. );
  308. var r = scheduler.CreateHotObservable(
  309. OnNext(150, 1),
  310. OnNext(250, 2),
  311. OnCompleted<int>(260)
  312. );
  313. var res = scheduler.Start(() =>
  314. l.TakeUntil(r.Do(_ => signalNotDisposed = true))
  315. );
  316. res.Messages.AssertEqual(
  317. OnNext(230, 2),
  318. OnCompleted<int>(240)
  319. );
  320. Assert.False(signalNotDisposed);
  321. }
  322. [Fact]
  323. public void TakeUntil_Error_Some()
  324. {
  325. var scheduler = new TestScheduler();
  326. var ex = new Exception();
  327. var l = scheduler.CreateHotObservable(
  328. OnNext(150, 1),
  329. OnError<int>(225, ex)
  330. );
  331. var r = scheduler.CreateHotObservable(
  332. OnNext(150, 1),
  333. OnNext(240, 2)
  334. );
  335. var res = scheduler.Start(() =>
  336. l.TakeUntil(r)
  337. );
  338. res.Messages.AssertEqual(
  339. OnError<int>(225, ex)
  340. );
  341. l.Subscriptions.AssertEqual(
  342. Subscribe(200, 225)
  343. );
  344. r.Subscriptions.AssertEqual(
  345. Subscribe(200, 225)
  346. );
  347. }
  348. [Fact]
  349. public void TakeUntil_Immediate()
  350. {
  351. var scheduler = new TestScheduler();
  352. var xs = Observable.Return(1);
  353. var ys = Observable.Return("bar");
  354. var res = scheduler.Start(() =>
  355. xs.TakeUntil(ys)
  356. );
  357. res.Messages.AssertEqual(
  358. OnCompleted<int>(200)
  359. );
  360. }
  361. #endregion
  362. #region + Timed +
  363. [Fact]
  364. public void TakeUntil_Timed_ArgumentChecking()
  365. {
  366. var xs = Observable.Return(42);
  367. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil(default(IObservable<int>), DateTimeOffset.Now));
  368. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil(default(IObservable<int>), DateTimeOffset.Now, Scheduler.Default));
  369. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil(xs, DateTimeOffset.Now, default));
  370. }
  371. [Fact]
  372. public void TakeUntil_Zero()
  373. {
  374. var scheduler = new TestScheduler();
  375. var xs = scheduler.CreateHotObservable(
  376. OnNext(210, 1),
  377. OnNext(220, 2),
  378. OnCompleted<int>(230)
  379. );
  380. var res = scheduler.Start(() =>
  381. xs.TakeUntil(new DateTimeOffset(), scheduler)
  382. );
  383. res.Messages.AssertEqual(
  384. OnCompleted<int>(201)
  385. );
  386. xs.Subscriptions.AssertEqual(
  387. Subscribe(200, 201)
  388. );
  389. }
  390. [Fact]
  391. public void TakeUntil_Some()
  392. {
  393. var scheduler = new TestScheduler();
  394. var xs = scheduler.CreateHotObservable(
  395. OnNext(210, 1),
  396. OnNext(220, 2),
  397. OnNext(230, 3),
  398. OnCompleted<int>(240)
  399. );
  400. var res = scheduler.Start(() =>
  401. xs.TakeUntil(new DateTimeOffset(225, TimeSpan.Zero), scheduler)
  402. );
  403. res.Messages.AssertEqual(
  404. OnNext(210, 1),
  405. OnNext(220, 2),
  406. OnCompleted<int>(225)
  407. );
  408. xs.Subscriptions.AssertEqual(
  409. Subscribe(200, 225)
  410. );
  411. }
  412. [Fact]
  413. public void TakeUntil_Late()
  414. {
  415. var scheduler = new TestScheduler();
  416. var xs = scheduler.CreateHotObservable(
  417. OnNext(210, 1),
  418. OnNext(220, 2),
  419. OnCompleted<int>(230)
  420. );
  421. var res = scheduler.Start(() =>
  422. xs.TakeUntil(new DateTimeOffset(250, TimeSpan.Zero), scheduler)
  423. );
  424. res.Messages.AssertEqual(
  425. OnNext(210, 1),
  426. OnNext(220, 2),
  427. OnCompleted<int>(230)
  428. );
  429. xs.Subscriptions.AssertEqual(
  430. Subscribe(200, 230)
  431. );
  432. }
  433. [Fact]
  434. public void TakeUntil_Error()
  435. {
  436. var scheduler = new TestScheduler();
  437. var ex = new Exception();
  438. var xs = scheduler.CreateHotObservable(
  439. OnError<int>(210, ex)
  440. );
  441. var res = scheduler.Start(() =>
  442. xs.TakeUntil(new DateTimeOffset(250, TimeSpan.Zero), scheduler)
  443. );
  444. res.Messages.AssertEqual(
  445. OnError<int>(210, ex)
  446. );
  447. xs.Subscriptions.AssertEqual(
  448. Subscribe(200, 210)
  449. );
  450. }
  451. [Fact]
  452. public void TakeUntil_Never()
  453. {
  454. var scheduler = new TestScheduler();
  455. var ex = new Exception();
  456. var xs = scheduler.CreateHotObservable<int>(
  457. );
  458. var res = scheduler.Start(() =>
  459. xs.TakeUntil(new DateTimeOffset(250, TimeSpan.Zero), scheduler)
  460. );
  461. res.Messages.AssertEqual(
  462. OnCompleted<int>(250)
  463. );
  464. xs.Subscriptions.AssertEqual(
  465. Subscribe(200, 250)
  466. );
  467. }
  468. [Fact]
  469. public void TakeUntil_Twice1()
  470. {
  471. var scheduler = new TestScheduler();
  472. var ex = new Exception();
  473. var xs = scheduler.CreateHotObservable(
  474. OnNext(210, 1),
  475. OnNext(220, 2),
  476. OnNext(230, 3),
  477. OnNext(240, 4),
  478. OnNext(250, 5),
  479. OnNext(260, 6),
  480. OnCompleted<int>(270)
  481. );
  482. var res = scheduler.Start(() =>
  483. xs.TakeUntil(new DateTimeOffset(255, TimeSpan.Zero), scheduler).TakeUntil(new DateTimeOffset(235, TimeSpan.Zero), scheduler)
  484. );
  485. res.Messages.AssertEqual(
  486. OnNext(210, 1),
  487. OnNext(220, 2),
  488. OnNext(230, 3),
  489. OnCompleted<int>(235)
  490. );
  491. xs.Subscriptions.AssertEqual(
  492. Subscribe(200, 235)
  493. );
  494. }
  495. [Fact]
  496. public void TakeUntil_Twice2()
  497. {
  498. var scheduler = new TestScheduler();
  499. var ex = new Exception();
  500. var xs = scheduler.CreateHotObservable(
  501. OnNext(210, 1),
  502. OnNext(220, 2),
  503. OnNext(230, 3),
  504. OnNext(240, 4),
  505. OnNext(250, 5),
  506. OnNext(260, 6),
  507. OnCompleted<int>(270)
  508. );
  509. var res = scheduler.Start(() =>
  510. xs.TakeUntil(new DateTimeOffset(235, TimeSpan.Zero), scheduler).TakeUntil(new DateTimeOffset(255, TimeSpan.Zero), scheduler)
  511. );
  512. res.Messages.AssertEqual(
  513. OnNext(210, 1),
  514. OnNext(220, 2),
  515. OnNext(230, 3),
  516. OnCompleted<int>(235)
  517. );
  518. xs.Subscriptions.AssertEqual(
  519. Subscribe(200, 235)
  520. );
  521. }
  522. [Fact]
  523. public void TakeUntil_Default()
  524. {
  525. var xs = Observable.Range(0, 10, Scheduler.Default);
  526. var res = xs.TakeUntil(DateTimeOffset.Now.AddMinutes(1));
  527. var e = new ManualResetEvent(false);
  528. var lst = new List<int>();
  529. res.Subscribe(
  530. lst.Add,
  531. () => e.Set()
  532. );
  533. e.WaitOne();
  534. Assert.True(lst.SequenceEqual(Enumerable.Range(0, 10)));
  535. }
  536. #endregion
  537. #region + Predicate +
  538. [Fact]
  539. public void TakeUntil_Predicate_ArgumentChecking()
  540. {
  541. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil<int>(null, v => true));
  542. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.TakeUntil(DummyObservable<int>.Instance, null));
  543. }
  544. [Fact]
  545. public void TakeUntil_Predicate_Basic()
  546. {
  547. var scheduler = new TestScheduler();
  548. var source = scheduler.CreateColdObservable(
  549. OnNext(10, 1),
  550. OnNext(20, 2),
  551. OnNext(30, 3),
  552. OnNext(40, 4),
  553. OnNext(50, 5),
  554. OnNext(60, 6),
  555. OnNext(70, 7),
  556. OnNext(80, 8),
  557. OnNext(90, 9),
  558. OnCompleted<int>(100)
  559. );
  560. var result = scheduler.Start(() => source.TakeUntil(v => v == 5));
  561. result.Messages.AssertEqual(
  562. OnNext(210, 1),
  563. OnNext(220, 2),
  564. OnNext(230, 3),
  565. OnNext(240, 4),
  566. OnNext(250, 5),
  567. OnCompleted<int>(250)
  568. );
  569. source.Subscriptions.AssertEqual(
  570. Subscribe(200, 250)
  571. );
  572. }
  573. [Fact]
  574. public void TakeUntil_Predicate_True()
  575. {
  576. var scheduler = new TestScheduler();
  577. var source = scheduler.CreateColdObservable(
  578. OnNext(10, 1),
  579. OnNext(20, 2),
  580. OnNext(30, 3),
  581. OnNext(40, 4),
  582. OnNext(50, 5),
  583. OnNext(60, 6),
  584. OnNext(70, 7),
  585. OnNext(80, 8),
  586. OnNext(90, 9),
  587. OnCompleted<int>(100)
  588. );
  589. var result = scheduler.Start(() => source.TakeUntil(v => true));
  590. result.Messages.AssertEqual(
  591. OnNext(210, 1),
  592. OnCompleted<int>(210)
  593. );
  594. source.Subscriptions.AssertEqual(
  595. Subscribe(200, 210)
  596. );
  597. }
  598. [Fact]
  599. public void TakeUntil_Predicate_False()
  600. {
  601. var scheduler = new TestScheduler();
  602. var source = scheduler.CreateColdObservable(
  603. OnNext(10, 1),
  604. OnNext(20, 2),
  605. OnNext(30, 3),
  606. OnNext(40, 4),
  607. OnNext(50, 5),
  608. OnNext(60, 6),
  609. OnNext(70, 7),
  610. OnNext(80, 8),
  611. OnNext(90, 9),
  612. OnCompleted<int>(100)
  613. );
  614. var result = scheduler.Start(() => source.TakeUntil(v => false));
  615. result.Messages.AssertEqual(
  616. OnNext(210, 1),
  617. OnNext(220, 2),
  618. OnNext(230, 3),
  619. OnNext(240, 4),
  620. OnNext(250, 5),
  621. OnNext(260, 6),
  622. OnNext(270, 7),
  623. OnNext(280, 8),
  624. OnNext(290, 9),
  625. OnCompleted<int>(300)
  626. );
  627. source.Subscriptions.AssertEqual(
  628. Subscribe(200, 300)
  629. );
  630. }
  631. [Fact]
  632. public void TakeUntil_Predicate_Error()
  633. {
  634. var scheduler = new TestScheduler();
  635. var ex = new InvalidOperationException();
  636. var source = scheduler.CreateColdObservable(
  637. OnNext(10, 1),
  638. OnNext(20, 2),
  639. OnNext(30, 3),
  640. OnError<int>(40, ex)
  641. );
  642. var result = scheduler.Start(() => source.TakeUntil(v => false));
  643. result.Messages.AssertEqual(
  644. OnNext(210, 1),
  645. OnNext(220, 2),
  646. OnNext(230, 3),
  647. OnError<int>(240, ex)
  648. );
  649. source.Subscriptions.AssertEqual(
  650. Subscribe(200, 240)
  651. );
  652. }
  653. [Fact]
  654. public void TakeUntil_Predicate_Crash()
  655. {
  656. var scheduler = new TestScheduler();
  657. var ex = new InvalidOperationException();
  658. var source = scheduler.CreateColdObservable(
  659. OnNext(10, 1),
  660. OnNext(20, 2),
  661. OnNext(30, 3),
  662. OnNext(240, 4),
  663. OnNext(250, 5),
  664. OnCompleted<int>(260)
  665. );
  666. var result = scheduler.Start(() => source.TakeUntil(v =>
  667. {
  668. if (v == 3)
  669. {
  670. throw ex;
  671. }
  672. return false;
  673. }));
  674. result.Messages.AssertEqual(
  675. OnNext(210, 1),
  676. OnNext(220, 2),
  677. OnNext(230, 3),
  678. OnError<int>(230, ex)
  679. );
  680. source.Subscriptions.AssertEqual(
  681. Subscribe(200, 230)
  682. );
  683. }
  684. #endregion
  685. }
  686. }