TaskObservableExtensionsTest.cs 28 KB

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