FromAsyncTest.cs 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  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.Linq;
  6. using System.Reactive;
  7. using System.Reactive.Concurrency;
  8. using System.Reactive.Linq;
  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 FromAsyncTest : ReactiveTest
  16. {
  17. private readonly Task<int> _doneTask;
  18. public FromAsyncTest()
  19. {
  20. var tcs = new TaskCompletionSource<int>();
  21. tcs.SetResult(42);
  22. _doneTask = tcs.Task;
  23. }
  24. #region Func
  25. [Fact]
  26. public void FromAsync_Func_ArgumentChecking()
  27. {
  28. var s = Scheduler.Immediate;
  29. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync<int>(default(Func<Task<int>>)));
  30. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync<int>(default(Func<CancellationToken, Task<int>>)));
  31. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync<int>(default(Func<Task<int>>), s));
  32. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync<int>(() => _doneTask, default));
  33. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync<int>(default(Func<CancellationToken, Task<int>>), s));
  34. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync<int>(ct => _doneTask, default));
  35. }
  36. [Fact]
  37. public void FromAsync_Func_Success()
  38. {
  39. var n = 42;
  40. var i = 0;
  41. var xs = Observable.FromAsync(() =>
  42. {
  43. i++;
  44. return Task.Factory.StartNew(() => n);
  45. });
  46. Assert.Equal(n, xs.Single());
  47. Assert.Equal(1, i);
  48. Assert.Equal(n, xs.Single());
  49. Assert.Equal(2, i);
  50. }
  51. [Fact]
  52. public void FromAsync_Func_Throw_Synchronous()
  53. {
  54. var ex = new Exception();
  55. var xs = Observable.FromAsync<int>(() =>
  56. {
  57. throw ex;
  58. });
  59. ReactiveAssert.Throws(ex, () => xs.Single());
  60. }
  61. [Fact]
  62. public void FromAsync_Func_Throw_Asynchronous()
  63. {
  64. var ex = new Exception();
  65. var xs = Observable.FromAsync<int>(() =>
  66. Task.Factory.StartNew<int>(() => { throw ex; })
  67. );
  68. ReactiveAssert.Throws(ex, () => xs.Single());
  69. }
  70. [Fact]
  71. public void FromAsync_FuncWithCancel_Success()
  72. {
  73. var n = 42;
  74. var i = 0;
  75. var xs = Observable.FromAsync(ct =>
  76. {
  77. i++;
  78. return Task.Factory.StartNew(() => n);
  79. });
  80. Assert.Equal(n, xs.Single());
  81. Assert.Equal(1, i);
  82. Assert.Equal(n, xs.Single());
  83. Assert.Equal(2, i);
  84. }
  85. [Fact]
  86. public void FromAsync_FuncWithCancel_Throw_Synchronous()
  87. {
  88. var ex = new Exception();
  89. var xs = Observable.FromAsync<int>(ct =>
  90. {
  91. throw ex;
  92. });
  93. ReactiveAssert.Throws(ex, () => xs.Single());
  94. }
  95. [Fact]
  96. public void FromAsync_FuncWithCancel_Throw_Asynchronous()
  97. {
  98. var ex = new Exception();
  99. var xs = Observable.FromAsync<int>(ct =>
  100. Task.Factory.StartNew<int>(() => { throw ex; })
  101. );
  102. ReactiveAssert.Throws(ex, () => xs.Single());
  103. }
  104. [Fact]
  105. public void FromAsync_FuncWithCancel_Cancel()
  106. {
  107. var e = new ManualResetEvent(false);
  108. var f = new ManualResetEvent(false);
  109. var t = default(Task<int>);
  110. var xs = Observable.FromAsync(ct =>
  111. t = Task.Factory.StartNew<int>(() =>
  112. {
  113. try
  114. {
  115. e.Set();
  116. while (true)
  117. {
  118. ct.ThrowIfCancellationRequested();
  119. }
  120. }
  121. finally
  122. {
  123. f.Set();
  124. }
  125. })
  126. );
  127. var d = xs.Subscribe(_ => { });
  128. e.WaitOne();
  129. d.Dispose();
  130. f.WaitOne();
  131. while (!t.IsCompleted)
  132. {
  133. ;
  134. }
  135. }
  136. #if DESKTOPCLR
  137. [Fact]
  138. public void FromAsync_Func_Scheduler1()
  139. {
  140. var e = new ManualResetEvent(false);
  141. var x = default(int);
  142. var t = default(int);
  143. var tcs = new TaskCompletionSource<int>();
  144. var xs = Observable.FromAsync(() => tcs.Task, Scheduler.Immediate);
  145. xs.Subscribe(res =>
  146. {
  147. x = res;
  148. t = Thread.CurrentThread.ManagedThreadId;
  149. e.Set();
  150. });
  151. tcs.SetResult(42);
  152. e.WaitOne();
  153. Assert.Equal(42, x);
  154. Assert.Equal(Thread.CurrentThread.ManagedThreadId, t);
  155. }
  156. [Fact]
  157. public void FromAsync_Func_Scheduler2()
  158. {
  159. var e = new ManualResetEvent(false);
  160. var x = default(int);
  161. var t = default(int);
  162. var tcs = new TaskCompletionSource<int>();
  163. var xs = Observable.FromAsync(ct => tcs.Task, Scheduler.Immediate);
  164. xs.Subscribe(res =>
  165. {
  166. x = res;
  167. t = Thread.CurrentThread.ManagedThreadId;
  168. e.Set();
  169. });
  170. tcs.SetResult(42);
  171. e.WaitOne();
  172. Assert.Equal(42, x);
  173. Assert.Equal(Thread.CurrentThread.ManagedThreadId, t);
  174. }
  175. #endif
  176. #endregion
  177. #region Action
  178. [Fact]
  179. public void FromAsync_Action_ArgumentChecking()
  180. {
  181. var s = Scheduler.Immediate;
  182. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync(default(Func<Task>)));
  183. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync(default(Func<CancellationToken, Task>)));
  184. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync(default(Func<Task>), s));
  185. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync(() => (Task)_doneTask, default));
  186. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync(default(Func<CancellationToken, Task>), s));
  187. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.FromAsync(ct => (Task)_doneTask, default));
  188. }
  189. [Fact]
  190. public void FromAsync_Action_Success()
  191. {
  192. var i = 0;
  193. var xs = Observable.FromAsync(() =>
  194. {
  195. i++;
  196. return Task.Factory.StartNew(() => { });
  197. });
  198. Assert.Equal(Unit.Default, xs.Single());
  199. Assert.Equal(1, i);
  200. Assert.Equal(Unit.Default, xs.Single());
  201. Assert.Equal(2, i);
  202. }
  203. [Fact]
  204. public void FromAsync_Action_Throw_Synchronous()
  205. {
  206. var ex = new Exception();
  207. var xs = Observable.FromAsync(() =>
  208. {
  209. throw ex;
  210. });
  211. ReactiveAssert.Throws(ex, () => xs.Single());
  212. }
  213. [Fact]
  214. public void FromAsync_Action_Throw_Asynchronous()
  215. {
  216. var ex = new Exception();
  217. var xs = Observable.FromAsync(() =>
  218. Task.Factory.StartNew(() => { throw ex; })
  219. );
  220. ReactiveAssert.Throws(ex, () => xs.Single());
  221. }
  222. [Fact]
  223. public void FromAsync_ActionWithCancel_Success()
  224. {
  225. var i = 0;
  226. var xs = Observable.FromAsync(ct =>
  227. {
  228. i++;
  229. return Task.Factory.StartNew(() => { });
  230. });
  231. Assert.Equal(Unit.Default, xs.Single());
  232. Assert.Equal(1, i);
  233. Assert.Equal(Unit.Default, xs.Single());
  234. Assert.Equal(2, i);
  235. }
  236. [Fact]
  237. public void FromAsync_ActionWithCancel_Throw_Synchronous()
  238. {
  239. var ex = new Exception();
  240. var xs = Observable.FromAsync(ct =>
  241. {
  242. throw ex;
  243. });
  244. ReactiveAssert.Throws(ex, () => xs.Single());
  245. }
  246. [Fact]
  247. public void FromAsync_ActionWithCancel_Throw_Asynchronous()
  248. {
  249. var ex = new Exception();
  250. var xs = Observable.FromAsync(ct =>
  251. Task.Factory.StartNew(() => { throw ex; })
  252. );
  253. ReactiveAssert.Throws(ex, () => xs.Single());
  254. }
  255. [Fact]
  256. public void FromAsync_ActionWithCancel_Cancel()
  257. {
  258. var e = new ManualResetEvent(false);
  259. var f = new ManualResetEvent(false);
  260. var t = default(Task);
  261. var xs = Observable.FromAsync(ct =>
  262. t = Task.Factory.StartNew(() =>
  263. {
  264. try
  265. {
  266. e.Set();
  267. while (true)
  268. {
  269. ct.ThrowIfCancellationRequested();
  270. }
  271. }
  272. finally
  273. {
  274. f.Set();
  275. }
  276. })
  277. );
  278. var d = xs.Subscribe(_ => { });
  279. e.WaitOne();
  280. d.Dispose();
  281. f.WaitOne();
  282. while (!t.IsCompleted)
  283. {
  284. ;
  285. }
  286. }
  287. #if DESKTOPCLR
  288. [Fact]
  289. public void FromAsync_Action_Scheduler1()
  290. {
  291. var e = new ManualResetEvent(false);
  292. var t = default(int);
  293. var tcs = new TaskCompletionSource<int>();
  294. var xs = Observable.FromAsync(() => (Task)tcs.Task, Scheduler.Immediate);
  295. xs.Subscribe(res =>
  296. {
  297. t = Thread.CurrentThread.ManagedThreadId;
  298. e.Set();
  299. });
  300. tcs.SetResult(42);
  301. e.WaitOne();
  302. Assert.Equal(Thread.CurrentThread.ManagedThreadId, t);
  303. }
  304. [Fact]
  305. public void FromAsync_Action_Scheduler2()
  306. {
  307. var e = new ManualResetEvent(false);
  308. var t = default(int);
  309. var tcs = new TaskCompletionSource<int>();
  310. var xs = Observable.FromAsync(ct => (Task)tcs.Task, Scheduler.Immediate);
  311. xs.Subscribe(res =>
  312. {
  313. t = Thread.CurrentThread.ManagedThreadId;
  314. e.Set();
  315. });
  316. tcs.SetResult(42);
  317. e.WaitOne();
  318. Assert.Equal(Thread.CurrentThread.ManagedThreadId, t);
  319. }
  320. #endif
  321. #endregion
  322. }
  323. }