SchedulerTest.cs 51 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518
  1. // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
  2. #if NET45
  3. using System;
  4. using System.Collections.Generic;
  5. using System.Linq;
  6. using System.Reactive.Concurrency;
  7. using System.Reactive.Disposables;
  8. using System.Reactive.Linq;
  9. using System.Reactive.PlatformServices;
  10. using System.Runtime.CompilerServices;
  11. using System.Threading;
  12. using Xunit;
  13. using Microsoft.Reactive.Testing;
  14. #if HAS_WINFORMS
  15. using System.Windows.Forms;
  16. #endif
  17. #if HAS_AWAIT
  18. using System.Threading.Tasks;
  19. #endif
  20. namespace ReactiveTests.Tests
  21. {
  22. public class SchedulerTest : ReactiveTest
  23. {
  24. #region IScheduler
  25. [Fact]
  26. public void Scheduler_ArgumentChecks()
  27. {
  28. var ms = new MyScheduler();
  29. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), a => { }));
  30. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), () => { }));
  31. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), 1, (a, s) => { }));
  32. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(ms, default(Action<Action>)));
  33. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(ms, 1, default(Action<int, Action<int>>)));
  34. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), DateTimeOffset.Now, a => { }));
  35. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), DateTimeOffset.Now, () => { }));
  36. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), 1, DateTimeOffset.Now, (a, s) => { }));
  37. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(ms, DateTimeOffset.Now, default(Action<Action<DateTimeOffset>>)));
  38. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(ms, 1, DateTimeOffset.Now, default(Action<int, Action<int, DateTimeOffset>>)));
  39. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), TimeSpan.Zero, a => { }));
  40. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), TimeSpan.Zero, () => { }));
  41. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(default(IScheduler), 1, TimeSpan.Zero, (a, s) => { }));
  42. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(ms, TimeSpan.Zero, default(Action<Action<TimeSpan>>)));
  43. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Schedule(ms, 1, TimeSpan.Zero, default(Action<int, Action<int, TimeSpan>>)));
  44. }
  45. [Fact]
  46. public void Schedulers_ArgumentChecks()
  47. {
  48. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.CurrentThread.Schedule(default(Action)));
  49. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.CurrentThread.Schedule(TimeSpan.Zero, default(Action)));
  50. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.CurrentThread.Schedule(DateTimeOffset.MaxValue, default(Action)));
  51. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherScheduler.Instance.Schedule(default(Action)));
  52. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherScheduler.Instance.Schedule(TimeSpan.Zero, default(Action)));
  53. ReactiveAssert.Throws<ArgumentNullException>(() => DispatcherScheduler.Instance.Schedule(DateTimeOffset.MaxValue, default(Action)));
  54. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Immediate.Schedule(default(Action)));
  55. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Immediate.Schedule(TimeSpan.Zero, default(Action)));
  56. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Immediate.Schedule(DateTimeOffset.MaxValue, default(Action)));
  57. ReactiveAssert.Throws<ArgumentNullException>(() => NewThreadScheduler.Default.Schedule(default(Action)));
  58. ReactiveAssert.Throws<ArgumentNullException>(() => NewThreadScheduler.Default.Schedule(TimeSpan.Zero, default(Action)));
  59. ReactiveAssert.Throws<ArgumentNullException>(() => NewThreadScheduler.Default.Schedule(DateTimeOffset.MaxValue, default(Action)));
  60. #if !NO_TPL
  61. ReactiveAssert.Throws<ArgumentNullException>(() => TaskPoolScheduler.Default.Schedule(default(Action)));
  62. ReactiveAssert.Throws<ArgumentNullException>(() => TaskPoolScheduler.Default.Schedule(TimeSpan.Zero, default(Action)));
  63. ReactiveAssert.Throws<ArgumentNullException>(() => TaskPoolScheduler.Default.Schedule(DateTimeOffset.MaxValue, default(Action)));
  64. #endif
  65. ReactiveAssert.Throws<ArgumentNullException>(() => DefaultScheduler.Instance.Schedule(default(Action)));
  66. ReactiveAssert.Throws<ArgumentNullException>(() => DefaultScheduler.Instance.Schedule(TimeSpan.Zero, default(Action)));
  67. ReactiveAssert.Throws<ArgumentNullException>(() => DefaultScheduler.Instance.Schedule(DateTimeOffset.MaxValue, default(Action)));
  68. ReactiveAssert.Throws<ArgumentNullException>(() => DefaultScheduler.Instance.SchedulePeriodic(42, TimeSpan.FromSeconds(1), default(Func<int, int>)));
  69. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => DefaultScheduler.Instance.SchedulePeriodic(42, TimeSpan.FromSeconds(-1), _ => _));
  70. #if HAS_WINFORMS
  71. var lbl = new Label();
  72. ReactiveAssert.Throws<ArgumentNullException>(() => new ControlScheduler(lbl).Schedule(default(Action)));
  73. ReactiveAssert.Throws<ArgumentNullException>(() => new ControlScheduler(lbl).Schedule(TimeSpan.Zero, default(Action)));
  74. ReactiveAssert.Throws<ArgumentNullException>(() => new ControlScheduler(lbl).Schedule(DateTimeOffset.MaxValue, default(Action)));
  75. #endif
  76. var ctx = new SynchronizationContext();
  77. ReactiveAssert.Throws<ArgumentNullException>(() => new SynchronizationContextScheduler(ctx).Schedule(default(Action)));
  78. ReactiveAssert.Throws<ArgumentNullException>(() => new SynchronizationContextScheduler(ctx).Schedule(TimeSpan.Zero, default(Action)));
  79. ReactiveAssert.Throws<ArgumentNullException>(() => new SynchronizationContextScheduler(ctx).Schedule(DateTimeOffset.MaxValue, default(Action)));
  80. }
  81. [Fact]
  82. public void Scheduler_ScheduleNonRecursive()
  83. {
  84. var ms = new MyScheduler();
  85. var res = false;
  86. Scheduler.Schedule(ms, a => { res = true; });
  87. Assert.True(res);
  88. }
  89. [Fact]
  90. public void Scheduler_ScheduleRecursive()
  91. {
  92. var ms = new MyScheduler();
  93. var i = 0;
  94. Scheduler.Schedule(ms, a => { if (++i < 10) a(); });
  95. Assert.Equal(10, i);
  96. }
  97. [Fact]
  98. public void Scheduler_ScheduleWithTimeNonRecursive()
  99. {
  100. var now = DateTimeOffset.Now;
  101. var ms = new MyScheduler(now) { Check = (a, s, t) => { Assert.True(t == TimeSpan.Zero); } };
  102. var res = false;
  103. Scheduler.Schedule(ms, now, a => { res = true; });
  104. Assert.True(res);
  105. Assert.True(ms.WaitCycles == 0);
  106. }
  107. [Fact]
  108. public void Scheduler_ScheduleWithTimeRecursive()
  109. {
  110. var now = DateTimeOffset.Now;
  111. var i = 0;
  112. var ms = new MyScheduler(now) { Check = (a, s, t) => { Assert.True(t == TimeSpan.Zero); } };
  113. Scheduler.Schedule(ms, now, a => { if (++i < 10) a(now); });
  114. Assert.True(ms.WaitCycles == 0);
  115. Assert.Equal(10, i);
  116. }
  117. [Fact]
  118. public void Scheduler_ScheduleWithTimeSpanNonRecursive()
  119. {
  120. var now = DateTimeOffset.Now;
  121. var ms = new MyScheduler(now) { Check = (a, s, t) => { Assert.True(t == TimeSpan.Zero); } };
  122. var res = false;
  123. Scheduler.Schedule(ms, TimeSpan.Zero, a => { res = true; });
  124. Assert.True(res);
  125. Assert.True(ms.WaitCycles == 0);
  126. }
  127. [Fact]
  128. public void Scheduler_ScheduleWithTimeSpanRecursive()
  129. {
  130. var now = DateTimeOffset.Now;
  131. var ms = new MyScheduler(now) { Check = (a, s, t) => { Assert.True(t < TimeSpan.FromTicks(10)); } };
  132. var i = 0;
  133. Scheduler.Schedule(ms, TimeSpan.Zero, a => { if (++i < 10) a(TimeSpan.FromTicks(i)); });
  134. Assert.True(ms.WaitCycles == Enumerable.Range(1, 9).Sum());
  135. Assert.Equal(10, i);
  136. }
  137. [Fact]
  138. public void Scheduler_StateThreading()
  139. {
  140. var lst = new List<int>();
  141. Scheduler.Immediate.Schedule(0, (i, a) =>
  142. {
  143. lst.Add(i);
  144. if (i < 9)
  145. a(i + 1);
  146. });
  147. Assert.True(lst.SequenceEqual(Enumerable.Range(0, 10)));
  148. }
  149. [Fact]
  150. public void Scheduler_Builtins()
  151. {
  152. // Default
  153. {
  154. var e = new ManualResetEvent(false);
  155. Scheduler.Default.Schedule(() => e.Set());
  156. e.WaitOne();
  157. }
  158. if (!Utils.IsRunningWithPortableLibraryBinaries())
  159. {
  160. Scheduler_Builtins_NoPlib();
  161. }
  162. }
  163. [MethodImpl(MethodImplOptions.NoInlining)]
  164. private void Scheduler_Builtins_NoPlib()
  165. {
  166. #if !PLIB
  167. // ThreadPool
  168. {
  169. var e = new ManualResetEvent(false);
  170. Scheduler.ThreadPool.Schedule(() => e.Set());
  171. e.WaitOne();
  172. }
  173. #endif
  174. #if !NO_THREAD
  175. // NewThread
  176. {
  177. var e = new ManualResetEvent(false);
  178. Scheduler.NewThread.Schedule(() => e.Set());
  179. e.WaitOne();
  180. }
  181. #endif
  182. #if !NO_TPL
  183. // TaskPool
  184. {
  185. var e = new ManualResetEvent(false);
  186. Scheduler.TaskPool.Schedule(() => e.Set());
  187. e.WaitOne();
  188. }
  189. #endif
  190. }
  191. #endregion
  192. #region ISchedulerLongRunning
  193. #if !NO_PERF
  194. [Fact]
  195. public void Scheduler_LongRunning_ArgumentChecking()
  196. {
  197. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleLongRunning(null, c => { }));
  198. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleLongRunning(ThreadPoolScheduler.Instance, default(Action<ICancelable>)));
  199. }
  200. [Fact]
  201. public void Scheduler_Periodic_ArgumentChecking()
  202. {
  203. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.SchedulePeriodic(null, TimeSpan.FromSeconds(1), () => { }));
  204. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Scheduler.SchedulePeriodic(ThreadPoolScheduler.Instance, TimeSpan.FromSeconds(-1), () => { }));
  205. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.SchedulePeriodic(ThreadPoolScheduler.Instance, TimeSpan.FromSeconds(1), default(Action)));
  206. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.SchedulePeriodic<int>(null, 42, TimeSpan.FromSeconds(1), _ => { }));
  207. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Scheduler.SchedulePeriodic<int>(ThreadPoolScheduler.Instance, 42, TimeSpan.FromSeconds(-1), _ => { }));
  208. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.SchedulePeriodic<int>(ThreadPoolScheduler.Instance, 42, TimeSpan.FromSeconds(1), default(Action<int>)));
  209. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.SchedulePeriodic<int>(null, 42, TimeSpan.FromSeconds(1), _ => _));
  210. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Scheduler.SchedulePeriodic<int>(ThreadPoolScheduler.Instance, 42, TimeSpan.FromSeconds(-1), _ => _));
  211. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.SchedulePeriodic<int>(ThreadPoolScheduler.Instance, 42, TimeSpan.FromSeconds(1), default(Func<int, int>)));
  212. }
  213. [Fact]
  214. public void Scheduler_Stopwatch_Emulation()
  215. {
  216. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.StartStopwatch(null));
  217. }
  218. #if !NO_TPL
  219. [Fact]
  220. public void Scheduler_LongRunning1()
  221. {
  222. var s = TaskPoolScheduler.Default;
  223. var x = new ManualResetEvent(false);
  224. var e = new ManualResetEvent(false);
  225. var d = s.ScheduleLongRunning(42, (state, cancel) =>
  226. {
  227. while (!cancel.IsDisposed)
  228. x.Set();
  229. e.Set();
  230. });
  231. x.WaitOne();
  232. d.Dispose();
  233. e.WaitOne();
  234. }
  235. [Fact]
  236. public void Scheduler_LongRunning2()
  237. {
  238. var s = TaskPoolScheduler.Default;
  239. var x = new ManualResetEvent(false);
  240. var e = new ManualResetEvent(false);
  241. var d = s.ScheduleLongRunning(cancel =>
  242. {
  243. while (!cancel.IsDisposed)
  244. x.Set();
  245. e.Set();
  246. });
  247. x.WaitOne();
  248. d.Dispose();
  249. e.WaitOne();
  250. }
  251. #endif
  252. #endif
  253. #endregion
  254. #region ISchedulerPeriodic
  255. #if !NO_PERF
  256. [Fact]
  257. public void Scheduler_Periodic1()
  258. {
  259. var n = 0;
  260. var e = new ManualResetEvent(false);
  261. var d = ThreadPoolScheduler.Instance.SchedulePeriodic(TimeSpan.FromMilliseconds(50), () =>
  262. {
  263. if (n++ == 10)
  264. e.Set();
  265. });
  266. e.WaitOne();
  267. d.Dispose();
  268. }
  269. [Fact]
  270. public void Scheduler_Periodic2()
  271. {
  272. var n = 0;
  273. var e = new ManualResetEvent(false);
  274. var d = ThreadPoolScheduler.Instance.SchedulePeriodic(42, TimeSpan.FromMilliseconds(50), x =>
  275. {
  276. Assert.Equal(42, x);
  277. if (n++ == 10)
  278. e.Set();
  279. });
  280. e.WaitOne();
  281. d.Dispose();
  282. }
  283. #if DESKTOPCLR
  284. [Fact]
  285. public void Scheduler_Periodic_HostLifecycleManagement()
  286. {
  287. var cur = AppDomain.CurrentDomain.BaseDirectory;
  288. var domain = AppDomain.CreateDomain("HLN", null, new AppDomainSetup { ApplicationBase = cur });
  289. domain.DoCallBack(Scheduler_Periodic_HostLifecycleManagement_Callback);
  290. }
  291. private static void Scheduler_Periodic_HostLifecycleManagement_Callback()
  292. {
  293. var pep = PlatformEnlightenmentProvider.Current;
  294. try
  295. {
  296. var hln = new HLN();
  297. PlatformEnlightenmentProvider.Current = new PEP(hln);
  298. var s = ThreadPoolScheduler.Instance.DisableOptimizations(typeof(ISchedulerPeriodic));
  299. var n = 0;
  300. var e = new ManualResetEvent(false);
  301. var d = Observable.Interval(TimeSpan.FromMilliseconds(100), s).Subscribe(_ =>
  302. {
  303. if (n++ == 10)
  304. e.Set();
  305. });
  306. hln.OnSuspending();
  307. hln.OnResuming();
  308. Thread.Sleep(250);
  309. hln.OnSuspending();
  310. Thread.Sleep(150);
  311. hln.OnResuming();
  312. Thread.Sleep(50);
  313. hln.OnSuspending();
  314. hln.OnResuming();
  315. e.WaitOne();
  316. d.Dispose();
  317. }
  318. finally
  319. {
  320. PlatformEnlightenmentProvider.Current = pep;
  321. }
  322. }
  323. class PEP : IPlatformEnlightenmentProvider
  324. {
  325. private readonly IPlatformEnlightenmentProvider _old;
  326. private readonly IHostLifecycleNotifications _hln;
  327. public PEP(HLN hln)
  328. {
  329. _old = PlatformEnlightenmentProvider.Current;
  330. _hln = hln;
  331. }
  332. public T GetService<T>(params object[] args) where T : class
  333. {
  334. if (typeof(T) == typeof(IHostLifecycleNotifications))
  335. {
  336. return (T)(object)_hln;
  337. }
  338. return _old.GetService<T>(args);
  339. }
  340. }
  341. class HLN : IHostLifecycleNotifications
  342. {
  343. public event EventHandler<HostSuspendingEventArgs> Suspending;
  344. public event EventHandler<HostResumingEventArgs> Resuming;
  345. public void OnSuspending()
  346. {
  347. var s = Suspending;
  348. if (s != null)
  349. s(this, null);
  350. }
  351. public void OnResuming()
  352. {
  353. var s = Resuming;
  354. if (s != null)
  355. s(this, null);
  356. }
  357. }
  358. #endif
  359. #endif
  360. #endregion
  361. #region DisableOptimizations
  362. #if !NO_PERF
  363. [Fact]
  364. public void DisableOptimizations_ArgumentChecking()
  365. {
  366. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.DisableOptimizations(default(IScheduler)));
  367. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.DisableOptimizations(default(IScheduler), new Type[0]));
  368. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.DisableOptimizations(ThreadPoolScheduler.Instance, default(Type[])));
  369. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.DisableOptimizations(Scheduler.Default).Schedule(42, default(Func<IScheduler, int, IDisposable>)));
  370. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.DisableOptimizations(Scheduler.Default).Schedule(42, TimeSpan.FromSeconds(1), default(Func<IScheduler, int, IDisposable>)));
  371. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.DisableOptimizations(Scheduler.Default).Schedule(42, DateTimeOffset.Now, default(Func<IScheduler, int, IDisposable>)));
  372. }
  373. #if !NO_TPL
  374. [Fact]
  375. public void DisableOptimizations1()
  376. {
  377. var s = TaskPoolScheduler.Default;
  378. Assert.True(s is IServiceProvider);
  379. var t = s.DisableOptimizations();
  380. var d = t.Now - s.Now;
  381. Assert.True(d.TotalSeconds < 1);
  382. var e1 = new ManualResetEvent(false);
  383. t.Schedule(42, (self, state) =>
  384. {
  385. e1.Set();
  386. return Disposable.Empty;
  387. });
  388. e1.WaitOne();
  389. var e2 = new ManualResetEvent(false);
  390. t.Schedule(42, TimeSpan.FromMilliseconds(1), (self, state) =>
  391. {
  392. e2.Set();
  393. return Disposable.Empty;
  394. });
  395. e2.WaitOne();
  396. var e3 = new ManualResetEvent(false);
  397. t.Schedule(42, DateTimeOffset.Now.AddMilliseconds(10), (self, state) =>
  398. {
  399. e3.Set();
  400. return Disposable.Empty;
  401. });
  402. e3.WaitOne();
  403. }
  404. [Fact]
  405. public void DisableOptimizations2()
  406. {
  407. var s = TaskPoolScheduler.Default;
  408. Assert.True(s is IServiceProvider);
  409. var lr1 = ((IServiceProvider)s).GetService(typeof(ISchedulerLongRunning));
  410. Assert.NotNull(lr1);
  411. var e1 = new ManualResetEvent(false);
  412. s.Schedule(42, (self, state) =>
  413. {
  414. Assert.True(self is IServiceProvider);
  415. var lrr1 = ((IServiceProvider)self).GetService(typeof(ISchedulerLongRunning));
  416. Assert.NotNull(lrr1);
  417. e1.Set();
  418. return Disposable.Empty;
  419. });
  420. e1.WaitOne();
  421. var t = s.DisableOptimizations();
  422. Assert.True(t is IServiceProvider);
  423. var lr2 = ((IServiceProvider)t).GetService(typeof(ISchedulerLongRunning));
  424. Assert.Null(lr2);
  425. var e2 = new ManualResetEvent(false);
  426. t.Schedule(42, (self, state) =>
  427. {
  428. Assert.True(self is IServiceProvider);
  429. var lrr2 = ((IServiceProvider)self).GetService(typeof(ISchedulerLongRunning));
  430. Assert.Null(lrr2);
  431. e2.Set();
  432. return Disposable.Empty;
  433. });
  434. e2.WaitOne();
  435. }
  436. [Fact]
  437. public void DisableOptimizations3()
  438. {
  439. var s = TaskPoolScheduler.Default;
  440. Assert.True(s is IServiceProvider);
  441. var lr1 = ((IServiceProvider)s).GetService(typeof(ISchedulerLongRunning));
  442. Assert.NotNull(lr1);
  443. var p1 = ((IServiceProvider)s).GetService(typeof(ISchedulerPeriodic));
  444. Assert.NotNull(p1);
  445. var e1 = new ManualResetEvent(false);
  446. s.Schedule(42, (self, state) =>
  447. {
  448. Assert.True(self is IServiceProvider);
  449. var lrr1 = ((IServiceProvider)self).GetService(typeof(ISchedulerLongRunning));
  450. Assert.NotNull(lrr1);
  451. var pr1 = ((IServiceProvider)self).GetService(typeof(ISchedulerPeriodic));
  452. Assert.NotNull(pr1);
  453. e1.Set();
  454. return Disposable.Empty;
  455. });
  456. e1.WaitOne();
  457. var t = s.DisableOptimizations(typeof(ISchedulerLongRunning));
  458. Assert.True(t is IServiceProvider);
  459. var lr2 = ((IServiceProvider)t).GetService(typeof(ISchedulerLongRunning));
  460. Assert.Null(lr2);
  461. var p2 = ((IServiceProvider)t).GetService(typeof(ISchedulerPeriodic));
  462. Assert.NotNull(p2);
  463. var e2 = new ManualResetEvent(false);
  464. t.Schedule(42, (self, state) =>
  465. {
  466. Assert.True(self is IServiceProvider);
  467. var lrr2 = ((IServiceProvider)self).GetService(typeof(ISchedulerLongRunning));
  468. Assert.Null(lrr2);
  469. var pr2 = ((IServiceProvider)self).GetService(typeof(ISchedulerPeriodic));
  470. Assert.NotNull(pr2);
  471. e2.Set();
  472. return Disposable.Empty;
  473. });
  474. e2.WaitOne();
  475. }
  476. #endif
  477. #endif
  478. [Fact]
  479. public void DisableOptimizations_UnknownService()
  480. {
  481. var s = new MyScheduler();
  482. Assert.False(s is IServiceProvider);
  483. var d = s.DisableOptimizations();
  484. Assert.True(d is IServiceProvider);
  485. Assert.Null(((IServiceProvider)d).GetService(typeof(bool)));
  486. }
  487. class MyScheduler : IScheduler
  488. {
  489. public MyScheduler()
  490. : this(DateTimeOffset.Now)
  491. {
  492. }
  493. public MyScheduler(DateTimeOffset now)
  494. {
  495. Now = now;
  496. }
  497. public DateTimeOffset Now
  498. {
  499. get;
  500. private set;
  501. }
  502. public IDisposable Schedule<T>(T state, Func<IScheduler, T, IDisposable> action)
  503. {
  504. return action(this, state);
  505. }
  506. public Action<Action<object>, object, TimeSpan> Check { get; set; }
  507. public long WaitCycles { get; set; }
  508. public IDisposable Schedule<T>(T state, TimeSpan dueTime, Func<IScheduler, T, IDisposable> action)
  509. {
  510. Check(o => action(this, (T)o), state, dueTime);
  511. WaitCycles += dueTime.Ticks;
  512. return action(this, state);
  513. }
  514. public IDisposable Schedule<T>(T state, DateTimeOffset dueTime, Func<IScheduler, T, IDisposable> action)
  515. {
  516. return Schedule(state, dueTime - Now, action);
  517. }
  518. }
  519. #endregion
  520. #region Catch
  521. [Fact]
  522. public void Catch_ArgumentChecking()
  523. {
  524. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Catch<Exception>(default(IScheduler), _ => true));
  525. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Catch<Exception>(Scheduler.Default, default(Func<Exception, bool>)));
  526. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Catch<Exception>(Scheduler.Default, _ => true).Schedule(42, default(Func<IScheduler, int, IDisposable>)));
  527. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Catch<Exception>(Scheduler.Default, _ => true).Schedule(42, TimeSpan.FromSeconds(1), default(Func<IScheduler, int, IDisposable>)));
  528. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Catch<Exception>(Scheduler.Default, _ => true).Schedule(42, DateTimeOffset.Now, default(Func<IScheduler, int, IDisposable>)));
  529. }
  530. [Fact]
  531. public void Catch_Builtin_Swallow_Shallow()
  532. {
  533. var done = new ManualResetEvent(false);
  534. var swallow = Scheduler.Default.Catch<InvalidOperationException>(_ => { done.Set(); return true; });
  535. swallow.Schedule(() =>
  536. {
  537. throw new InvalidOperationException();
  538. });
  539. done.WaitOne();
  540. }
  541. [Fact]
  542. public void Catch_Builtin_Swallow_Recursive()
  543. {
  544. var done = new ManualResetEvent(false);
  545. var swallow = Scheduler.Default.Catch<InvalidOperationException>(_ => { done.Set(); return true; });
  546. swallow.Schedule(42, (self, state) =>
  547. {
  548. return self.Schedule(() =>
  549. {
  550. throw new InvalidOperationException();
  551. });
  552. });
  553. done.WaitOne();
  554. }
  555. [Fact]
  556. public void Catch_Custom_Unhandled()
  557. {
  558. var err = default(Exception);
  559. var scheduler = new MyExceptionScheduler(ex_ => err = ex_);
  560. scheduler.Catch<InvalidOperationException>(_ => true).Schedule(() =>
  561. {
  562. throw new InvalidOperationException();
  563. });
  564. Assert.Null(err);
  565. var ex = new ArgumentException();
  566. scheduler.Catch<InvalidOperationException>(_ => true).Schedule(() =>
  567. {
  568. throw ex;
  569. });
  570. Assert.Same(ex, err);
  571. }
  572. [Fact]
  573. public void Catch_Custom_Rethrow()
  574. {
  575. var err = default(Exception);
  576. var scheduler = new MyExceptionScheduler(ex_ => err = ex_);
  577. var ex = new InvalidOperationException();
  578. scheduler.Catch<InvalidOperationException>(_ => false).Schedule(() =>
  579. {
  580. throw ex;
  581. });
  582. Assert.Same(ex, err);
  583. }
  584. [Fact]
  585. public void Catch_Custom_LongRunning_Caught()
  586. {
  587. var err = default(Exception);
  588. var scheduler = new MyExceptionScheduler(ex_ => err = ex_);
  589. var caught = false;
  590. var @catch = scheduler.Catch<InvalidOperationException>(_ => caught = true);
  591. var slr = (ISchedulerLongRunning)((IServiceProvider)@catch).GetService(typeof(ISchedulerLongRunning));
  592. slr.ScheduleLongRunning(cancel =>
  593. {
  594. throw new InvalidOperationException();
  595. });
  596. Assert.True(caught);
  597. Assert.Null(err);
  598. var ex = new ArgumentException();
  599. slr.ScheduleLongRunning(cancel =>
  600. {
  601. throw ex;
  602. });
  603. Assert.Same(ex, err);
  604. }
  605. [Fact]
  606. public void Catch_Custom_LongRunning_Rethrow()
  607. {
  608. var err = default(Exception);
  609. var scheduler = new MyExceptionScheduler(ex_ => err = ex_);
  610. var @catch = scheduler.Catch<InvalidOperationException>(_ => false);
  611. var slr = (ISchedulerLongRunning)((IServiceProvider)@catch).GetService(typeof(ISchedulerLongRunning));
  612. var done = false;
  613. slr.ScheduleLongRunning(cancel =>
  614. {
  615. done = true;
  616. });
  617. Assert.True(done);
  618. var ex = new InvalidOperationException();
  619. slr.ScheduleLongRunning(cancel =>
  620. {
  621. throw ex;
  622. });
  623. Assert.Same(ex, err);
  624. }
  625. [Fact]
  626. public void Catch_Custom_Periodic_Regular()
  627. {
  628. var scheduler = new MyExceptionScheduler(_ => { });
  629. scheduler.PeriodicStopped = new ManualResetEvent(false);
  630. var @catch = scheduler.Catch<InvalidOperationException>(_ => true);
  631. var per = (ISchedulerPeriodic)((IServiceProvider)@catch).GetService(typeof(ISchedulerPeriodic));
  632. var madeProgress = new ManualResetEvent(false);
  633. var d = per.SchedulePeriodic(0, TimeSpan.Zero, x =>
  634. {
  635. if (x > 10)
  636. madeProgress.Set();
  637. return x + 1;
  638. });
  639. madeProgress.WaitOne();
  640. d.Dispose();
  641. scheduler.PeriodicStopped.WaitOne();
  642. }
  643. [Fact]
  644. public void Catch_Custom_Periodic_Uncaught1()
  645. {
  646. var err = default(Exception);
  647. var done = new ManualResetEvent(false);
  648. var scheduler = new MyExceptionScheduler(ex_ => { err = ex_; done.Set(); });
  649. scheduler.PeriodicStopped = new ManualResetEvent(false);
  650. var @catch = scheduler.Catch<InvalidOperationException>(_ => true);
  651. var per = (ISchedulerPeriodic)((IServiceProvider)@catch).GetService(typeof(ISchedulerPeriodic));
  652. var ex = new ArgumentException();
  653. per.SchedulePeriodic(42, TimeSpan.Zero, x =>
  654. {
  655. throw ex;
  656. });
  657. scheduler.PeriodicStopped.WaitOne();
  658. done.WaitOne();
  659. Assert.Same(ex, err);
  660. }
  661. [Fact]
  662. public void Catch_Custom_Periodic_Uncaught2()
  663. {
  664. var err = default(Exception);
  665. var done = new ManualResetEvent(false);
  666. var scheduler = new MyExceptionScheduler(ex_ => { err = ex_; done.Set(); });
  667. scheduler.PeriodicStopped = new ManualResetEvent(false);
  668. var @catch = scheduler.Catch<InvalidOperationException>(_ => false);
  669. var per = (ISchedulerPeriodic)((IServiceProvider)@catch).GetService(typeof(ISchedulerPeriodic));
  670. var ex = new InvalidOperationException();
  671. per.SchedulePeriodic(42, TimeSpan.Zero, x =>
  672. {
  673. throw ex;
  674. });
  675. scheduler.PeriodicStopped.WaitOne();
  676. done.WaitOne();
  677. Assert.Same(ex, err);
  678. }
  679. [Fact]
  680. public void Catch_Custom_Periodic_Caught()
  681. {
  682. var err = default(Exception);
  683. var scheduler = new MyExceptionScheduler(ex_ => err = ex_);
  684. scheduler.PeriodicStopped = new ManualResetEvent(false);
  685. var caught = new ManualResetEvent(false);
  686. var @catch = scheduler.Catch<InvalidOperationException>(_ => { caught.Set(); return true; });
  687. var per = (ISchedulerPeriodic)((IServiceProvider)@catch).GetService(typeof(ISchedulerPeriodic));
  688. per.SchedulePeriodic(42, TimeSpan.Zero, x =>
  689. {
  690. throw new InvalidOperationException();
  691. });
  692. scheduler.PeriodicStopped.WaitOne();
  693. caught.WaitOne();
  694. Assert.Null(err);
  695. }
  696. class MyExceptionScheduler : LocalScheduler, ISchedulerLongRunning, ISchedulerPeriodic
  697. {
  698. private readonly Action<Exception> _onError;
  699. public MyExceptionScheduler(Action<Exception> onError)
  700. {
  701. _onError = onError;
  702. }
  703. public override IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
  704. {
  705. try
  706. {
  707. return action(this, state);
  708. }
  709. catch (Exception exception)
  710. {
  711. _onError(exception);
  712. return Disposable.Empty;
  713. }
  714. }
  715. public override IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  716. {
  717. throw new NotImplementedException();
  718. }
  719. public IDisposable ScheduleLongRunning<TState>(TState state, Action<TState, ICancelable> action)
  720. {
  721. var b = new BooleanDisposable();
  722. try
  723. {
  724. action(state, b);
  725. }
  726. catch (Exception exception)
  727. {
  728. _onError(exception);
  729. return Disposable.Empty;
  730. }
  731. return b;
  732. }
  733. public ManualResetEvent PeriodicStopped { get; set; }
  734. public IDisposable SchedulePeriodic<TState>(TState state, TimeSpan period, Func<TState, TState> action)
  735. {
  736. var b = new BooleanDisposable();
  737. Scheduler.Default.Schedule(() =>
  738. {
  739. try
  740. {
  741. var s = state;
  742. for (int i = 0; true; i++)
  743. {
  744. if (i > 100 /* mimic delayed cancellation */ && b.IsDisposed)
  745. break;
  746. s = action(s);
  747. }
  748. }
  749. catch (Exception exception)
  750. {
  751. _onError(exception);
  752. }
  753. finally
  754. {
  755. PeriodicStopped.Set();
  756. }
  757. });
  758. return b;
  759. }
  760. }
  761. #endregion
  762. #region Services
  763. [Fact]
  764. public void InvalidService_Null()
  765. {
  766. var s = new MySchedulerWithoutServices();
  767. Assert.Null(((IServiceProvider)s).GetService(typeof(IAsyncResult)));
  768. }
  769. class MySchedulerWithoutServices : LocalScheduler
  770. {
  771. public override IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  772. {
  773. throw new NotImplementedException();
  774. }
  775. }
  776. [Fact]
  777. public void DetectServices_Null_1()
  778. {
  779. var s = new MyDumbScheduler1();
  780. Assert.Null(Scheduler.AsLongRunning(s));
  781. Assert.Null(Scheduler.AsPeriodic(s));
  782. Assert.Null(Scheduler.AsStopwatchProvider(s));
  783. }
  784. class MyDumbScheduler1 : IScheduler
  785. {
  786. public DateTimeOffset Now
  787. {
  788. get { throw new NotImplementedException(); }
  789. }
  790. public IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
  791. {
  792. throw new NotImplementedException();
  793. }
  794. public IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  795. {
  796. throw new NotImplementedException();
  797. }
  798. public IDisposable Schedule<TState>(TState state, DateTimeOffset dueTime, Func<IScheduler, TState, IDisposable> action)
  799. {
  800. throw new NotImplementedException();
  801. }
  802. }
  803. [Fact]
  804. public void DetectServices_Null_2()
  805. {
  806. var s = new MyDumbScheduler2(new Dictionary<Type, object>());
  807. Assert.Null(Scheduler.AsLongRunning(s));
  808. Assert.Null(Scheduler.AsPeriodic(s));
  809. Assert.Null(Scheduler.AsStopwatchProvider(s));
  810. }
  811. [Fact]
  812. public void DetectServices_Found()
  813. {
  814. {
  815. var x = new MyLongRunning();
  816. var s = new MyDumbScheduler2(new Dictionary<Type, object>
  817. {
  818. { typeof(ISchedulerLongRunning), x }
  819. });
  820. Assert.Equal(x, Scheduler.AsLongRunning(s));
  821. }
  822. {
  823. var x = new MyStopwatchProvider();
  824. var s = new MyDumbScheduler2(new Dictionary<Type, object>
  825. {
  826. { typeof(IStopwatchProvider), x }
  827. });
  828. Assert.Equal(x, Scheduler.AsStopwatchProvider(s));
  829. }
  830. {
  831. var x = new MyPeriodic();
  832. var s = new MyDumbScheduler2(new Dictionary<Type, object>
  833. {
  834. { typeof(ISchedulerPeriodic), x }
  835. });
  836. Assert.Equal(x, Scheduler.AsPeriodic(s));
  837. }
  838. }
  839. class MyDumbScheduler2 : IScheduler, IServiceProvider
  840. {
  841. private readonly Dictionary<Type, object> _services;
  842. public MyDumbScheduler2(Dictionary<Type, object> services)
  843. {
  844. _services = services;
  845. }
  846. public DateTimeOffset Now
  847. {
  848. get { throw new NotImplementedException(); }
  849. }
  850. public IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
  851. {
  852. throw new NotImplementedException();
  853. }
  854. public IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action)
  855. {
  856. throw new NotImplementedException();
  857. }
  858. public IDisposable Schedule<TState>(TState state, DateTimeOffset dueTime, Func<IScheduler, TState, IDisposable> action)
  859. {
  860. throw new NotImplementedException();
  861. }
  862. public object GetService(Type serviceType)
  863. {
  864. var res = default(object);
  865. if (_services.TryGetValue(serviceType, out res))
  866. return res;
  867. return null;
  868. }
  869. }
  870. class MyLongRunning : ISchedulerLongRunning
  871. {
  872. public IDisposable ScheduleLongRunning<TState>(TState state, Action<TState, ICancelable> action)
  873. {
  874. throw new NotImplementedException();
  875. }
  876. }
  877. class MyStopwatchProvider : IStopwatchProvider
  878. {
  879. public IStopwatch StartStopwatch()
  880. {
  881. throw new NotImplementedException();
  882. }
  883. }
  884. class MyPeriodic : ISchedulerPeriodic
  885. {
  886. public IDisposable SchedulePeriodic<TState>(TState state, TimeSpan period, Func<TState, TState> action)
  887. {
  888. throw new NotImplementedException();
  889. }
  890. }
  891. #endregion
  892. #if HAS_AWAIT
  893. [Fact]
  894. public void SchedulerAsync_Yield_ArgumentChecking()
  895. {
  896. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Yield(default(IScheduler)));
  897. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Yield(default(IScheduler), CancellationToken.None));
  898. }
  899. [Fact]
  900. public void SchedulerAsync_Sleep_ArgumentChecking()
  901. {
  902. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Sleep(default(IScheduler), TimeSpan.Zero));
  903. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Sleep(default(IScheduler), TimeSpan.Zero, CancellationToken.None));
  904. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Sleep(default(IScheduler), DateTimeOffset.MinValue));
  905. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.Sleep(default(IScheduler), DateTimeOffset.MinValue, CancellationToken.None));
  906. }
  907. [Fact]
  908. public void SchedulerAsync_ScheduleAsync_ArgumentChecking()
  909. {
  910. var tcs = new TaskCompletionSource<IDisposable>();
  911. var t = tcs.Task;
  912. var d = default(IScheduler);
  913. var s = Scheduler.Immediate;
  914. var rt = TimeSpan.Zero;
  915. var at = DateTimeOffset.MinValue;
  916. var a1 = new Func<IScheduler, int, CancellationToken, Task>((_, __, ___) => t);
  917. var d1 = default(Func<IScheduler, int, CancellationToken, Task>);
  918. var a2 = new Func<IScheduler, int, CancellationToken, Task<IDisposable>>((_, __, ___) => t);
  919. var d2 = default(Func<IScheduler, int, CancellationToken, Task<IDisposable>>);
  920. var a3 = new Func<IScheduler, CancellationToken, Task>((_, __) => t);
  921. var d3 = default(Func<IScheduler, CancellationToken, Task>);
  922. var a4 = new Func<IScheduler, CancellationToken, Task<IDisposable>>((_, __) => t);
  923. var d4 = default(Func<IScheduler, CancellationToken, Task<IDisposable>>);
  924. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, 42, a1));
  925. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, 42, d1));
  926. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, 42, rt, a1));
  927. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, 42, rt, d1));
  928. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, 42, at, a1));
  929. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, 42, at, d1));
  930. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, 42, a2));
  931. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, 42, d2));
  932. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, 42, rt, a2));
  933. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, 42, rt, d2));
  934. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, 42, at, a2));
  935. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, 42, at, d2));
  936. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, a3));
  937. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, d3));
  938. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, rt, a3));
  939. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, rt, d3));
  940. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, at, a3));
  941. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, at, d3));
  942. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, a4));
  943. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, d4));
  944. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, rt, a4));
  945. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, rt, d4));
  946. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(d, at, a4));
  947. ReactiveAssert.Throws<ArgumentNullException>(() => Scheduler.ScheduleAsync(s, at, d4));
  948. }
  949. [Fact]
  950. public void SchedulerAsync_ScheduleAsync_Overloads1()
  951. {
  952. var tcsI = new TaskCompletionSource<int>();
  953. var t = tcsI.Task;
  954. tcsI.SetResult(0);
  955. var tcsD = new TaskCompletionSource<IDisposable>();
  956. var d = tcsD.Task;
  957. tcsD.SetResult(Disposable.Empty);
  958. var s = new TestScheduler();
  959. var o = s.CreateObserver<int>();
  960. s.ScheduleAsync((_, ct) =>
  961. {
  962. o.OnNext(42);
  963. return t;
  964. });
  965. s.ScheduleAsync((_, ct) =>
  966. {
  967. o.OnNext(43);
  968. return d;
  969. });
  970. s.ScheduleAsync(44, (_, x, ct) =>
  971. {
  972. o.OnNext(x);
  973. return t;
  974. });
  975. s.ScheduleAsync(45, (_, x, ct) =>
  976. {
  977. o.OnNext(45);
  978. return d;
  979. });
  980. s.Start();
  981. o.Messages.AssertEqual(
  982. OnNext(1, 42),
  983. OnNext(1, 43),
  984. OnNext(1, 44),
  985. OnNext(1, 45)
  986. );
  987. }
  988. [Fact]
  989. public void SchedulerAsync_ScheduleAsync_Overloads2()
  990. {
  991. var tcsI = new TaskCompletionSource<int>();
  992. var t = tcsI.Task;
  993. tcsI.SetResult(0);
  994. var tcsD = new TaskCompletionSource<IDisposable>();
  995. var d = tcsD.Task;
  996. tcsD.SetResult(Disposable.Empty);
  997. var s = new TestScheduler();
  998. var o = s.CreateObserver<int>();
  999. s.ScheduleAsync(TimeSpan.FromTicks(50), (_, ct) =>
  1000. {
  1001. o.OnNext(42);
  1002. return t;
  1003. });
  1004. s.ScheduleAsync(TimeSpan.FromTicks(60), (_, ct) =>
  1005. {
  1006. o.OnNext(43);
  1007. return d;
  1008. });
  1009. s.ScheduleAsync(44, TimeSpan.FromTicks(70), (_, x, ct) =>
  1010. {
  1011. o.OnNext(x);
  1012. return t;
  1013. });
  1014. s.ScheduleAsync(45, TimeSpan.FromTicks(80), (_, x, ct) =>
  1015. {
  1016. o.OnNext(45);
  1017. return d;
  1018. });
  1019. s.Start();
  1020. o.Messages.AssertEqual(
  1021. OnNext(50, 42),
  1022. OnNext(60, 43),
  1023. OnNext(70, 44),
  1024. OnNext(80, 45)
  1025. );
  1026. }
  1027. [Fact]
  1028. public void SchedulerAsync_ScheduleAsync_Overloads3()
  1029. {
  1030. var tcsI = new TaskCompletionSource<int>();
  1031. var t = tcsI.Task;
  1032. tcsI.SetResult(0);
  1033. var tcsD = new TaskCompletionSource<IDisposable>();
  1034. var d = tcsD.Task;
  1035. tcsD.SetResult(Disposable.Empty);
  1036. var s = new TestScheduler();
  1037. var o = s.CreateObserver<int>();
  1038. s.ScheduleAsync(new DateTimeOffset(50, TimeSpan.Zero), (_, ct) =>
  1039. {
  1040. o.OnNext(42);
  1041. return t;
  1042. });
  1043. s.ScheduleAsync(new DateTimeOffset(60, TimeSpan.Zero), (_, ct) =>
  1044. {
  1045. o.OnNext(43);
  1046. return d;
  1047. });
  1048. s.ScheduleAsync(44, new DateTimeOffset(70, TimeSpan.Zero), (_, x, ct) =>
  1049. {
  1050. o.OnNext(x);
  1051. return t;
  1052. });
  1053. s.ScheduleAsync(45, new DateTimeOffset(80, TimeSpan.Zero), (_, x, ct) =>
  1054. {
  1055. o.OnNext(45);
  1056. return d;
  1057. });
  1058. s.Start();
  1059. o.Messages.AssertEqual(
  1060. OnNext(50, 42),
  1061. OnNext(60, 43),
  1062. OnNext(70, 44),
  1063. OnNext(80, 45)
  1064. );
  1065. }
  1066. [Fact]
  1067. public void SchedulerAsync_ScheduleAsync_NoCancellation1()
  1068. {
  1069. var s = new TestScheduler();
  1070. var o = s.CreateObserver<int>();
  1071. s.ScheduleAsync(async (_, ct) =>
  1072. {
  1073. o.OnNext(42);
  1074. await _.Yield();
  1075. o.OnNext(43);
  1076. await _.Sleep(TimeSpan.FromTicks(10));
  1077. o.OnNext(44);
  1078. await _.Sleep(new DateTimeOffset(250, TimeSpan.Zero));
  1079. o.OnNext(45);
  1080. });
  1081. s.Start();
  1082. o.Messages.AssertEqual(
  1083. OnNext(1, 42),
  1084. OnNext(2, 43),
  1085. OnNext(12, 44),
  1086. OnNext(250, 45)
  1087. );
  1088. }
  1089. [Fact]
  1090. public void SchedulerAsync_ScheduleAsync_NoCancellation2()
  1091. {
  1092. var s = new TestScheduler();
  1093. var o = s.CreateObserver<int>();
  1094. s.ScheduleAsync(async (_, ct) =>
  1095. {
  1096. o.OnNext(42);
  1097. await _.Yield(ct);
  1098. o.OnNext(43);
  1099. await _.Sleep(TimeSpan.FromTicks(10), ct);
  1100. o.OnNext(44);
  1101. await _.Sleep(new DateTimeOffset(250, TimeSpan.Zero), ct);
  1102. o.OnNext(45);
  1103. });
  1104. s.Start();
  1105. o.Messages.AssertEqual(
  1106. OnNext(1, 42),
  1107. OnNext(2, 43),
  1108. OnNext(12, 44),
  1109. OnNext(250, 45)
  1110. );
  1111. }
  1112. [Fact]
  1113. public void SchedulerAsync_Awaiters()
  1114. {
  1115. var op = Scheduler.Immediate.Yield();
  1116. var aw = op.GetAwaiter();
  1117. ReactiveAssert.Throws<ArgumentNullException>(() => aw.OnCompleted(null));
  1118. aw.OnCompleted(() => { });
  1119. ReactiveAssert.Throws<InvalidOperationException>(() => aw.OnCompleted(() => { }));
  1120. }
  1121. [Fact]
  1122. public void SchedulerAsync_Yield_Cancel1()
  1123. {
  1124. var cts = new CancellationTokenSource();
  1125. var op = Scheduler.Immediate.Yield(cts.Token);
  1126. var aw = op.GetAwaiter();
  1127. cts.Cancel();
  1128. Assert.True(aw.IsCompleted);
  1129. ReactiveAssert.Throws<OperationCanceledException>(() => aw.GetResult());
  1130. }
  1131. [Fact]
  1132. public void SchedulerAsync_Yield_Cancel2()
  1133. {
  1134. var cts = new CancellationTokenSource();
  1135. var op = Scheduler.Immediate.Yield(cts.Token);
  1136. var aw = op.GetAwaiter();
  1137. Assert.False(aw.IsCompleted);
  1138. aw.OnCompleted(() =>
  1139. {
  1140. //
  1141. // TODO: Not needed for await pattern, but maybe should be wired up?
  1142. //
  1143. // Assert.True(aw.IsCompleted);
  1144. cts.Cancel();
  1145. ReactiveAssert.Throws<OperationCanceledException>(() => aw.GetResult());
  1146. });
  1147. }
  1148. [Fact]
  1149. public void SchedulerAsync_Sleep_Cancel()
  1150. {
  1151. var cts = new CancellationTokenSource();
  1152. var op = Scheduler.Default.Sleep(TimeSpan.FromHours(1), cts.Token);
  1153. var aw = op.GetAwaiter();
  1154. Assert.False(aw.IsCompleted);
  1155. var e = new ManualResetEvent(false);
  1156. aw.OnCompleted(() =>
  1157. {
  1158. ReactiveAssert.Throws<OperationCanceledException>(() => aw.GetResult());
  1159. e.Set();
  1160. });
  1161. cts.Cancel();
  1162. e.WaitOne();
  1163. }
  1164. #if !NO_SYNCCTX
  1165. [Fact]
  1166. public void SchedulerAsync_ScheduleAsync_SyncCtx()
  1167. {
  1168. var old = SynchronizationContext.Current;
  1169. try
  1170. {
  1171. var ctx = new MySyncCtx();
  1172. SynchronizationContext.SetSynchronizationContext(ctx);
  1173. var s = new TestScheduler();
  1174. var o = s.CreateObserver<int>();
  1175. s.ScheduleAsync(async (_, ct) =>
  1176. {
  1177. Assert.Same(ctx, SynchronizationContext.Current);
  1178. o.OnNext(42);
  1179. await _.Yield(ct).ConfigureAwait(true);
  1180. Assert.Same(ctx, SynchronizationContext.Current);
  1181. o.OnNext(43);
  1182. await _.Sleep(TimeSpan.FromTicks(10), ct).ConfigureAwait(true);
  1183. Assert.Same(ctx, SynchronizationContext.Current);
  1184. o.OnNext(44);
  1185. await _.Sleep(new DateTimeOffset(250, TimeSpan.Zero), ct).ConfigureAwait(true);
  1186. Assert.Same(ctx, SynchronizationContext.Current);
  1187. o.OnNext(45);
  1188. });
  1189. s.Start();
  1190. o.Messages.AssertEqual(
  1191. OnNext(1, 42),
  1192. OnNext(2, 43),
  1193. OnNext(12, 44),
  1194. OnNext(250, 45)
  1195. );
  1196. }
  1197. finally
  1198. {
  1199. SynchronizationContext.SetSynchronizationContext(old);
  1200. }
  1201. }
  1202. class MySyncCtx : SynchronizationContext
  1203. {
  1204. public override void Post(SendOrPostCallback d, object state)
  1205. {
  1206. d(state);
  1207. }
  1208. }
  1209. #endif
  1210. #endif
  1211. }
  1212. }
  1213. #endif