SubscribeOnTest.cs 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355
  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.Reactive.Concurrency;
  6. using System.Reactive.Linq;
  7. using System.Threading;
  8. using Microsoft.Reactive.Testing;
  9. using ReactiveTests.Dummies;
  10. using Xunit;
  11. namespace ReactiveTests.Tests
  12. {
  13. public class SubscribeOnTest : TestBase
  14. {
  15. #region + TestBase +
  16. [Fact]
  17. public void SubscribeOn_ArgumentChecking()
  18. {
  19. var someObservable = Observable.Empty<int>();
  20. #if HAS_WINFORMS
  21. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn<int>(default(IObservable<int>), new ControlScheduler(new Label())));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn<int>(someObservable, default(ControlScheduler)));
  23. ReactiveAssert.Throws<ArgumentNullException>(() => ControlObservable.SubscribeOn<int>(default(IObservable<int>), new Label()));
  24. ReactiveAssert.Throws<ArgumentNullException>(() => ControlObservable.SubscribeOn<int>(someObservable, default(Label)));
  25. #endif
  26. #if HAS_DISPATCHER
  27. #if USE_SL_DISPATCHER
  28. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn<int>(default(IObservable<int>), new DispatcherScheduler(System.Windows.Deployment.Current.Dispatcher)));
  29. #else
  30. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn<int>(default(IObservable<int>), new DispatcherScheduler(Dispatcher.CurrentDispatcher)));
  31. #endif
  32. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn<int>(someObservable, default(DispatcherScheduler)));
  33. #if USE_SL_DISPATCHER
  34. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherObservable.SubscribeOn<int>(default(IObservable<int>), System.Windows.Deployment.Current.Dispatcher));
  35. #else
  36. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherObservable.SubscribeOn<int>(default(IObservable<int>), Dispatcher.CurrentDispatcher));
  37. #endif
  38. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherObservable.SubscribeOn<int>(someObservable, default(Dispatcher)));
  39. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherObservable.SubscribeOnDispatcher<int>(default(IObservable<int>)));
  40. #endif
  41. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn<int>(default, new SynchronizationContext()));
  42. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn(someObservable, default(SynchronizationContext)));
  43. }
  44. #if HAS_WINFORMS
  45. [Fact]
  46. public void SubscribeOn_Control()
  47. {
  48. var lbl = CreateLabel();
  49. var evt2 = new ManualResetEvent(false);
  50. var evt = new ManualResetEvent(false);
  51. bool okay = true;
  52. var d = Observable.Create<int>(obs =>
  53. {
  54. lbl.Text = "Subscribe";
  55. okay &= (SynchronizationContext.Current is System.Windows.Forms.WindowsFormsSynchronizationContext);
  56. evt2.Set();
  57. return () =>
  58. {
  59. lbl.Text = "Unsubscribe";
  60. okay &= (SynchronizationContext.Current is System.Windows.Forms.WindowsFormsSynchronizationContext);
  61. evt.Set();
  62. };
  63. })
  64. .SubscribeOn(lbl)
  65. .Subscribe(_ => {});
  66. evt2.WaitOne();
  67. d.Dispose();
  68. evt.WaitOne();
  69. Application.Exit();
  70. Assert.True(okay);
  71. }
  72. [Fact]
  73. public void SubscribeOn_ControlScheduler()
  74. {
  75. var lbl = CreateLabel();
  76. var evt2 = new ManualResetEvent(false);
  77. var evt = new ManualResetEvent(false);
  78. bool okay = true;
  79. var d = Observable.Create<int>(obs =>
  80. {
  81. lbl.Text = "Subscribe";
  82. okay &= (SynchronizationContext.Current is System.Windows.Forms.WindowsFormsSynchronizationContext);
  83. evt2.Set();
  84. return () =>
  85. {
  86. lbl.Text = "Unsubscribe";
  87. okay &= (SynchronizationContext.Current is System.Windows.Forms.WindowsFormsSynchronizationContext);
  88. evt.Set();
  89. };
  90. })
  91. .SubscribeOn(new ControlScheduler(lbl))
  92. .Subscribe(_ => { });
  93. evt2.WaitOne();
  94. d.Dispose();
  95. evt.WaitOne();
  96. Application.Exit();
  97. Assert.True(okay);
  98. }
  99. #endif
  100. #if HAS_DISPATCHER
  101. [Fact]
  102. [Asynchronous]
  103. public void SubscribeOn_Dispatcher()
  104. {
  105. var dispatcher = DispatcherHelpers.EnsureDispatcher();
  106. RunAsync(evt =>
  107. {
  108. var s = new AsyncSubject<Unit>();
  109. bool okay = true;
  110. var d = Observable.Create<int>(obs =>
  111. {
  112. okay &= (SynchronizationContext.Current is System.Windows.Threading.DispatcherSynchronizationContext);
  113. s.OnNext(Unit.Default);
  114. s.OnCompleted();
  115. return () =>
  116. {
  117. okay &= (SynchronizationContext.Current is System.Windows.Threading.DispatcherSynchronizationContext);
  118. Assert.True(okay);
  119. dispatcher.InvokeShutdown();
  120. evt.Set();
  121. };
  122. })
  123. .SubscribeOn(dispatcher)
  124. .Subscribe(_ => { });
  125. s.Subscribe(_ => d.Dispose());
  126. });
  127. }
  128. [Fact]
  129. [Asynchronous]
  130. public void SubscribeOn_DispatcherScheduler()
  131. {
  132. var dispatcher = DispatcherHelpers.EnsureDispatcher();
  133. RunAsync(evt =>
  134. {
  135. var s = new AsyncSubject<Unit>();
  136. bool okay = true;
  137. var d = Observable.Create<int>(obs =>
  138. {
  139. okay &= (SynchronizationContext.Current is System.Windows.Threading.DispatcherSynchronizationContext);
  140. s.OnNext(Unit.Default);
  141. s.OnCompleted();
  142. return () =>
  143. {
  144. okay &= (SynchronizationContext.Current is System.Windows.Threading.DispatcherSynchronizationContext);
  145. Assert.True(okay);
  146. dispatcher.InvokeShutdown();
  147. evt.Set();
  148. };
  149. })
  150. .SubscribeOn(new DispatcherScheduler(dispatcher))
  151. .Subscribe(_ => { });
  152. s.Subscribe(_ => d.Dispose());
  153. });
  154. }
  155. [Fact]
  156. [Asynchronous]
  157. public void SubscribeOn_CurrentDispatcher()
  158. {
  159. var dispatcher = DispatcherHelpers.EnsureDispatcher();
  160. RunAsync(evt =>
  161. {
  162. var s = new AsyncSubject<Unit>();
  163. bool okay = true;
  164. dispatcher.BeginInvoke(new Action(() =>
  165. {
  166. var d = Observable.Create<int>(obs =>
  167. {
  168. okay &= (SynchronizationContext.Current is System.Windows.Threading.DispatcherSynchronizationContext);
  169. s.OnNext(Unit.Default);
  170. s.OnCompleted();
  171. return () =>
  172. {
  173. okay &= (SynchronizationContext.Current is System.Windows.Threading.DispatcherSynchronizationContext);
  174. Assert.True(okay);
  175. dispatcher.InvokeShutdown();
  176. evt.Set();
  177. };
  178. })
  179. .SubscribeOnDispatcher()
  180. .Subscribe(_ => { });
  181. s.Subscribe(_ => d.Dispose());
  182. }));
  183. });
  184. }
  185. #endif
  186. #endregion + TestBase +
  187. }
  188. public class SubscribeOnReactiveTest : ReactiveTest
  189. {
  190. [Fact]
  191. public void SubscribeOn_Scheduler_ArgumentChecking()
  192. {
  193. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn(default(IObservable<int>), DummyScheduler.Instance));
  194. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.SubscribeOn(DummyObservable<int>.Instance, default(IScheduler)));
  195. }
  196. [Fact]
  197. public void SubscribeOn_Scheduler_Sleep()
  198. {
  199. var scheduler = new TestScheduler();
  200. var s = 0L;
  201. var d = 0L;
  202. var xs = Observable.Create<long>(observer =>
  203. {
  204. s = scheduler.Clock;
  205. return () => d = scheduler.Clock;
  206. });
  207. var results = scheduler.Start(() =>
  208. xs.SubscribeOn(scheduler)
  209. );
  210. results.Messages.AssertEqual(
  211. );
  212. Assert.Equal(201, s);
  213. Assert.Equal(1001, d);
  214. }
  215. [Fact]
  216. public void SubscribeOn_Scheduler_Completed()
  217. {
  218. var scheduler = new TestScheduler();
  219. var xs = scheduler.CreateHotObservable(
  220. OnCompleted<long>(300)
  221. );
  222. var results = scheduler.Start(() =>
  223. xs.SubscribeOn(scheduler)
  224. );
  225. results.Messages.AssertEqual(
  226. OnCompleted<long>(300)
  227. );
  228. xs.Subscriptions.AssertEqual(
  229. Subscribe(201, 301)
  230. );
  231. }
  232. [Fact]
  233. public void SubscribeOn_Scheduler_Error()
  234. {
  235. var scheduler = new TestScheduler();
  236. var ex = new Exception();
  237. var xs = scheduler.CreateHotObservable(
  238. OnError<int>(300, ex)
  239. );
  240. var results = scheduler.Start(() =>
  241. xs.SubscribeOn(scheduler)
  242. );
  243. results.Messages.AssertEqual(
  244. OnError<int>(300, ex)
  245. );
  246. xs.Subscriptions.AssertEqual(
  247. Subscribe(201, 301)
  248. );
  249. }
  250. [Fact]
  251. public void SubscribeOn_Scheduler_Dispose()
  252. {
  253. var scheduler = new TestScheduler();
  254. var xs = scheduler.CreateHotObservable<int>(
  255. );
  256. var results = scheduler.Start(() =>
  257. xs.SubscribeOn(scheduler)
  258. );
  259. results.Messages.AssertEqual(
  260. );
  261. xs.Subscriptions.AssertEqual(
  262. Subscribe(201, 1001)
  263. );
  264. }
  265. [Fact]
  266. public void SubscribeOn_SynchronizationContext_Simple()
  267. {
  268. var scheduler = new TestScheduler();
  269. var xs = scheduler.CreateHotObservable(
  270. OnNext(90, 1),
  271. OnNext(120, 2),
  272. OnNext(230, 3),
  273. OnNext(240, 4),
  274. OnNext(310, 5),
  275. OnNext(470, 6),
  276. OnCompleted<int>(530)
  277. );
  278. var results = scheduler.Start(() =>
  279. xs.SubscribeOn(new MyCtx(scheduler))
  280. );
  281. results.Messages.AssertEqual(
  282. OnNext(230, 3),
  283. OnNext(240, 4),
  284. OnNext(310, 5),
  285. OnNext(470, 6),
  286. OnCompleted<int>(530)
  287. );
  288. xs.Subscriptions.AssertEqual(
  289. Subscribe(201, 531)
  290. );
  291. }
  292. }
  293. }