TaskObservableExtensionsTest.cs 28 KB

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