ThrottleTest.cs 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656
  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.Collections.Generic;
  6. using System.Linq;
  7. using System.Reactive;
  8. using System.Reactive.Linq;
  9. using Microsoft.Reactive.Testing;
  10. using ReactiveTests.Dummies;
  11. using Xunit;
  12. namespace ReactiveTests.Tests
  13. {
  14. public class ThrottleTest : ReactiveTest
  15. {
  16. [Fact]
  17. public void Throttle_ArgumentChecking()
  18. {
  19. var scheduler = new TestScheduler();
  20. var someObservable = Observable.Empty<int>();
  21. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Throttle(default(IObservable<int>), TimeSpan.Zero));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Throttle(someObservable, TimeSpan.Zero, null));
  23. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Throttle(default(IObservable<int>), TimeSpan.Zero, scheduler));
  24. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Observable.Throttle(someObservable, TimeSpan.FromSeconds(-1)));
  25. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => Observable.Throttle(someObservable, TimeSpan.FromSeconds(-1), scheduler));
  26. }
  27. private IEnumerable<Recorded<Notification<T>>> Generate<T, S>(S seed, Func<S, bool> condition, Func<S, S> iterate, Func<S, Recorded<Notification<T>>> selector, Func<S, Recorded<Notification<T>>> final)
  28. {
  29. S s;
  30. for (s = seed; condition(s); s = iterate(s))
  31. {
  32. yield return selector(s);
  33. }
  34. yield return final(s);
  35. }
  36. [Fact]
  37. public void Throttle_TimeSpan_AllPass()
  38. {
  39. var scheduler = new TestScheduler();
  40. var xs = scheduler.CreateHotObservable(
  41. OnNext(150, 0),
  42. OnNext(210, 1),
  43. OnNext(240, 2),
  44. OnNext(270, 3),
  45. OnNext(300, 4),
  46. OnCompleted<int>(400)
  47. );
  48. var res = scheduler.Start(() =>
  49. xs.Throttle(TimeSpan.FromTicks(20), scheduler)
  50. );
  51. res.Messages.AssertEqual(
  52. OnNext(230, 1),
  53. OnNext(260, 2),
  54. OnNext(290, 3),
  55. OnNext(320, 4),
  56. OnCompleted<int>(400)
  57. );
  58. xs.Subscriptions.AssertEqual(
  59. Subscribe(200, 400)
  60. );
  61. }
  62. [Fact]
  63. public void Throttle_TimeSpan_AllPass_ErrorEnd()
  64. {
  65. var ex = new Exception();
  66. var scheduler = new TestScheduler();
  67. var xs = scheduler.CreateHotObservable(
  68. OnNext(150, 0),
  69. OnNext(210, 1),
  70. OnNext(240, 2),
  71. OnNext(270, 3),
  72. OnNext(300, 4),
  73. OnError<int>(400, ex)
  74. );
  75. var res = scheduler.Start(() =>
  76. xs.Throttle(TimeSpan.FromTicks(20), scheduler)
  77. );
  78. res.Messages.AssertEqual(
  79. OnNext(230, 1),
  80. OnNext(260, 2),
  81. OnNext(290, 3),
  82. OnNext(320, 4),
  83. OnError<int>(400, ex)
  84. );
  85. xs.Subscriptions.AssertEqual(
  86. Subscribe(200, 400)
  87. );
  88. }
  89. [Fact]
  90. public void Throttle_TimeSpan_AllDrop()
  91. {
  92. var scheduler = new TestScheduler();
  93. var xs = scheduler.CreateHotObservable(
  94. OnNext(150, 0),
  95. OnNext(210, 1),
  96. OnNext(240, 2),
  97. OnNext(270, 3),
  98. OnNext(300, 4),
  99. OnNext(330, 5),
  100. OnNext(360, 6),
  101. OnNext(390, 7),
  102. OnCompleted<int>(400)
  103. );
  104. var res = scheduler.Start(() =>
  105. xs.Throttle(TimeSpan.FromTicks(40), scheduler)
  106. );
  107. res.Messages.AssertEqual(
  108. OnNext(400, 7),
  109. OnCompleted<int>(400)
  110. );
  111. xs.Subscriptions.AssertEqual(
  112. Subscribe(200, 400)
  113. );
  114. }
  115. [Fact]
  116. public void Throttle_TimeSpan_AllDrop_ErrorEnd()
  117. {
  118. var ex = new Exception();
  119. var scheduler = new TestScheduler();
  120. var xs = scheduler.CreateHotObservable(
  121. OnNext(150, 0),
  122. OnNext(210, 1),
  123. OnNext(240, 2),
  124. OnNext(270, 3),
  125. OnNext(300, 4),
  126. OnNext(330, 5),
  127. OnNext(360, 6),
  128. OnNext(390, 7),
  129. OnError<int>(400, ex)
  130. );
  131. var res = scheduler.Start(() =>
  132. xs.Throttle(TimeSpan.FromTicks(40), scheduler)
  133. );
  134. res.Messages.AssertEqual(
  135. OnError<int>(400, ex)
  136. );
  137. xs.Subscriptions.AssertEqual(
  138. Subscribe(200, 400)
  139. );
  140. }
  141. [Fact]
  142. public void Throttle_Empty()
  143. {
  144. var scheduler = new TestScheduler();
  145. var xs = scheduler.CreateHotObservable(
  146. OnNext(150, 0),
  147. OnCompleted<int>(300)
  148. );
  149. var res = scheduler.Start(() =>
  150. xs.Throttle(TimeSpan.FromTicks(10), scheduler)
  151. );
  152. res.Messages.AssertEqual(
  153. OnCompleted<int>(300)
  154. );
  155. xs.Subscriptions.AssertEqual(
  156. Subscribe(200, 300)
  157. );
  158. }
  159. [Fact]
  160. public void Throttle_Error()
  161. {
  162. var scheduler = new TestScheduler();
  163. var ex = new Exception();
  164. var xs = scheduler.CreateHotObservable(
  165. OnNext(150, 0),
  166. OnError<int>(300, ex)
  167. );
  168. var res = scheduler.Start(() =>
  169. xs.Throttle(TimeSpan.FromTicks(10), scheduler)
  170. );
  171. res.Messages.AssertEqual(
  172. OnError<int>(300, ex)
  173. );
  174. xs.Subscriptions.AssertEqual(
  175. Subscribe(200, 300)
  176. );
  177. }
  178. [Fact]
  179. public void Throttle_Never()
  180. {
  181. var scheduler = new TestScheduler();
  182. var xs = scheduler.CreateHotObservable(
  183. OnNext(150, 0)
  184. );
  185. var res = scheduler.Start(() =>
  186. xs.Throttle(TimeSpan.FromTicks(10), scheduler)
  187. );
  188. res.Messages.AssertEqual(
  189. );
  190. xs.Subscriptions.AssertEqual(
  191. Subscribe(200, 1000)
  192. );
  193. }
  194. [Fact]
  195. public void Throttle_Simple()
  196. {
  197. var scheduler = new TestScheduler();
  198. var xs = scheduler.CreateHotObservable(
  199. OnNext(150, 0),
  200. OnNext(210, 1),
  201. OnNext(240, 2),
  202. OnNext(250, 3),
  203. OnNext(280, 4),
  204. OnCompleted<int>(300)
  205. );
  206. var res = scheduler.Start(() =>
  207. xs.Throttle(TimeSpan.FromTicks(20), scheduler)
  208. );
  209. res.Messages.AssertEqual(
  210. OnNext(230, 1),
  211. OnNext(270, 3),
  212. OnNext(300, 4),
  213. OnCompleted<int>(300)
  214. );
  215. xs.Subscriptions.AssertEqual(
  216. Subscribe(200, 300)
  217. );
  218. }
  219. [Fact]
  220. public void Throttle_DefaultScheduler()
  221. {
  222. Assert.True(Observable.Return(1).Throttle(TimeSpan.FromMilliseconds(1)).ToEnumerable().SequenceEqual(new[] { 1 }));
  223. }
  224. [Fact]
  225. public void Throttle_Duration_ArgumentChecking()
  226. {
  227. var someObservable = DummyObservable<int>.Instance;
  228. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Throttle(default(IObservable<int>), x => someObservable));
  229. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Throttle(someObservable, default(Func<int, IObservable<string>>)));
  230. }
  231. [Fact]
  232. public void Throttle_Duration_DelayBehavior()
  233. {
  234. var scheduler = new TestScheduler();
  235. var xs = scheduler.CreateHotObservable(
  236. OnNext(150, -1),
  237. OnNext(250, 0),
  238. OnNext(280, 1),
  239. OnNext(310, 2),
  240. OnNext(350, 3),
  241. OnNext(400, 4),
  242. OnCompleted<int>(550)
  243. );
  244. var ys = new[] {
  245. scheduler.CreateColdObservable(
  246. OnNext(20, 42),
  247. OnNext(25, 99)
  248. ),
  249. scheduler.CreateColdObservable(
  250. OnNext(20, 42),
  251. OnNext(25, 99)
  252. ),
  253. scheduler.CreateColdObservable(
  254. OnNext(20, 42),
  255. OnNext(25, 99)
  256. ),
  257. scheduler.CreateColdObservable(
  258. OnNext(20, 42),
  259. OnNext(25, 99)
  260. ),
  261. scheduler.CreateColdObservable(
  262. OnNext(20, 42),
  263. OnNext(25, 99)
  264. ),
  265. };
  266. var res = scheduler.Start(() =>
  267. xs.Throttle(x => ys[x])
  268. );
  269. res.Messages.AssertEqual(
  270. OnNext(250 + 20, 0),
  271. OnNext(280 + 20, 1),
  272. OnNext(310 + 20, 2),
  273. OnNext(350 + 20, 3),
  274. OnNext(400 + 20, 4),
  275. OnCompleted<int>(550)
  276. );
  277. xs.Subscriptions.AssertEqual(
  278. Subscribe(200, 550)
  279. );
  280. ys[0].Subscriptions.AssertEqual(Subscribe(250, 250 + 20));
  281. ys[1].Subscriptions.AssertEqual(Subscribe(280, 280 + 20));
  282. ys[2].Subscriptions.AssertEqual(Subscribe(310, 310 + 20));
  283. ys[3].Subscriptions.AssertEqual(Subscribe(350, 350 + 20));
  284. ys[4].Subscriptions.AssertEqual(Subscribe(400, 400 + 20));
  285. }
  286. [Fact]
  287. public void Throttle_Duration_ThrottleBehavior()
  288. {
  289. var scheduler = new TestScheduler();
  290. var xs = scheduler.CreateHotObservable(
  291. OnNext(150, -1),
  292. OnNext(250, 0),
  293. OnNext(280, 1),
  294. OnNext(310, 2),
  295. OnNext(350, 3),
  296. OnNext(400, 4),
  297. OnCompleted<int>(550)
  298. );
  299. var ys = new[] {
  300. scheduler.CreateColdObservable(
  301. OnNext(20, 42),
  302. OnNext(25, 99)
  303. ),
  304. scheduler.CreateColdObservable(
  305. OnNext(40, 42),
  306. OnNext(45, 99)
  307. ),
  308. scheduler.CreateColdObservable(
  309. OnNext(20, 42),
  310. OnNext(25, 99)
  311. ),
  312. scheduler.CreateColdObservable(
  313. OnNext(60, 42),
  314. OnNext(65, 99)
  315. ),
  316. scheduler.CreateColdObservable(
  317. OnNext(20, 42),
  318. OnNext(25, 99)
  319. ),
  320. };
  321. var res = scheduler.Start(() =>
  322. xs.Throttle(x => ys[x])
  323. );
  324. res.Messages.AssertEqual(
  325. OnNext(250 + 20, 0),
  326. OnNext(310 + 20, 2),
  327. OnNext(400 + 20, 4),
  328. OnCompleted<int>(550)
  329. );
  330. xs.Subscriptions.AssertEqual(
  331. Subscribe(200, 550)
  332. );
  333. ys[0].Subscriptions.AssertEqual(Subscribe(250, 250 + 20));
  334. ys[1].Subscriptions.AssertEqual(Subscribe(280, 310));
  335. ys[2].Subscriptions.AssertEqual(Subscribe(310, 310 + 20));
  336. ys[3].Subscriptions.AssertEqual(Subscribe(350, 400));
  337. ys[4].Subscriptions.AssertEqual(Subscribe(400, 400 + 20));
  338. }
  339. [Fact]
  340. public void Throttle_Duration_EarlyCompletion()
  341. {
  342. var scheduler = new TestScheduler();
  343. var xs = scheduler.CreateHotObservable(
  344. OnNext(150, -1),
  345. OnNext(250, 0),
  346. OnNext(280, 1),
  347. OnNext(310, 2),
  348. OnNext(350, 3),
  349. OnNext(400, 4),
  350. OnCompleted<int>(410)
  351. );
  352. var ys = new[] {
  353. scheduler.CreateColdObservable(
  354. OnNext(20, 42),
  355. OnNext(25, 99)
  356. ),
  357. scheduler.CreateColdObservable(
  358. OnNext(40, 42),
  359. OnNext(45, 99)
  360. ),
  361. scheduler.CreateColdObservable(
  362. OnNext(20, 42),
  363. OnNext(25, 99)
  364. ),
  365. scheduler.CreateColdObservable(
  366. OnNext(60, 42),
  367. OnNext(65, 99)
  368. ),
  369. scheduler.CreateColdObservable(
  370. OnNext(20, 42),
  371. OnNext(25, 99)
  372. ),
  373. };
  374. var res = scheduler.Start(() =>
  375. xs.Throttle(x => ys[x])
  376. );
  377. res.Messages.AssertEqual(
  378. OnNext(250 + 20, 0),
  379. OnNext(310 + 20, 2),
  380. OnNext(410, 4),
  381. OnCompleted<int>(410)
  382. );
  383. xs.Subscriptions.AssertEqual(
  384. Subscribe(200, 410)
  385. );
  386. ys[0].Subscriptions.AssertEqual(Subscribe(250, 250 + 20));
  387. ys[1].Subscriptions.AssertEqual(Subscribe(280, 310));
  388. ys[2].Subscriptions.AssertEqual(Subscribe(310, 310 + 20));
  389. ys[3].Subscriptions.AssertEqual(Subscribe(350, 400));
  390. ys[4].Subscriptions.AssertEqual(Subscribe(400, 410));
  391. }
  392. [Fact]
  393. public void Throttle_Duration_InnerError()
  394. {
  395. var scheduler = new TestScheduler();
  396. var xs = scheduler.CreateHotObservable(
  397. OnNext(150, 1),
  398. OnNext(250, 2),
  399. OnNext(350, 3),
  400. OnNext(450, 4),
  401. OnCompleted<int>(550)
  402. );
  403. var ex = new Exception();
  404. var res = scheduler.Start(() =>
  405. xs.Throttle(x =>
  406. x < 4 ? scheduler.CreateColdObservable(
  407. OnNext(x * 10, "Ignore"),
  408. OnNext(x * 10 + 5, "Aargh!")
  409. )
  410. : scheduler.CreateColdObservable(
  411. OnError<string>(x * 10, ex)
  412. )
  413. )
  414. );
  415. res.Messages.AssertEqual(
  416. OnNext(250 + 2 * 10, 2),
  417. OnNext(350 + 3 * 10, 3),
  418. OnError<int>(450 + 4 * 10, ex)
  419. );
  420. xs.Subscriptions.AssertEqual(
  421. Subscribe(200, 490)
  422. );
  423. }
  424. [Fact]
  425. public void Throttle_Duration_OuterError()
  426. {
  427. var scheduler = new TestScheduler();
  428. var ex = new Exception();
  429. var xs = scheduler.CreateHotObservable(
  430. OnNext(150, 1),
  431. OnNext(250, 2),
  432. OnNext(350, 3),
  433. OnNext(450, 4),
  434. OnError<int>(460, ex)
  435. );
  436. var res = scheduler.Start(() =>
  437. xs.Throttle(x =>
  438. scheduler.CreateColdObservable(
  439. OnNext(x * 10, "Ignore"),
  440. OnNext(x * 10 + 5, "Aargh!")
  441. )
  442. )
  443. );
  444. res.Messages.AssertEqual(
  445. OnNext(250 + 2 * 10, 2),
  446. OnNext(350 + 3 * 10, 3),
  447. OnError<int>(460, ex)
  448. );
  449. xs.Subscriptions.AssertEqual(
  450. Subscribe(200, 460)
  451. );
  452. }
  453. [Fact]
  454. public void Throttle_Duration_SelectorThrows()
  455. {
  456. var scheduler = new TestScheduler();
  457. var xs = scheduler.CreateHotObservable(
  458. OnNext(150, 1),
  459. OnNext(250, 2),
  460. OnNext(350, 3),
  461. OnNext(450, 4),
  462. OnCompleted<int>(550)
  463. );
  464. var ex = new Exception();
  465. var res = scheduler.Start(() =>
  466. xs.Throttle(x =>
  467. {
  468. if (x < 4)
  469. {
  470. return scheduler.CreateColdObservable(
  471. OnNext(x * 10, "Ignore"),
  472. OnNext(x * 10 + 5, "Aargh!")
  473. );
  474. }
  475. else
  476. {
  477. throw ex;
  478. }
  479. })
  480. );
  481. res.Messages.AssertEqual(
  482. OnNext(250 + 2 * 10, 2),
  483. OnNext(350 + 3 * 10, 3),
  484. OnError<int>(450, ex)
  485. );
  486. xs.Subscriptions.AssertEqual(
  487. Subscribe(200, 450)
  488. );
  489. }
  490. [Fact]
  491. public void Throttle_Duration_InnerDone_DelayBehavior()
  492. {
  493. var scheduler = new TestScheduler();
  494. var xs = scheduler.CreateHotObservable(
  495. OnNext(150, 1),
  496. OnNext(250, 2),
  497. OnNext(350, 3),
  498. OnNext(450, 4),
  499. OnCompleted<int>(550)
  500. );
  501. var ex = new Exception();
  502. var res = scheduler.Start(() =>
  503. xs.Throttle(x =>
  504. scheduler.CreateColdObservable(
  505. OnCompleted<string>(x * 10)
  506. )
  507. )
  508. );
  509. res.Messages.AssertEqual(
  510. OnNext(250 + 2 * 10, 2),
  511. OnNext(350 + 3 * 10, 3),
  512. OnNext(450 + 4 * 10, 4),
  513. OnCompleted<int>(550)
  514. );
  515. xs.Subscriptions.AssertEqual(
  516. Subscribe(200, 550)
  517. );
  518. }
  519. [Fact]
  520. public void Throttle_Duration_InnerDone_ThrottleBehavior()
  521. {
  522. var scheduler = new TestScheduler();
  523. var xs = scheduler.CreateHotObservable(
  524. OnNext(150, 1),
  525. OnNext(250, 2),
  526. OnNext(280, 3),
  527. OnNext(300, 4),
  528. OnNext(400, 5),
  529. OnNext(410, 6),
  530. OnCompleted<int>(550)
  531. );
  532. var ex = new Exception();
  533. var res = scheduler.Start(() =>
  534. xs.Throttle(x =>
  535. scheduler.CreateColdObservable(
  536. OnCompleted<string>(x * 10)
  537. )
  538. )
  539. );
  540. res.Messages.AssertEqual(
  541. OnNext(250 + 2 * 10, 2),
  542. OnNext(300 + 4 * 10, 4),
  543. OnNext(410 + 6 * 10, 6),
  544. OnCompleted<int>(550)
  545. );
  546. xs.Subscriptions.AssertEqual(
  547. Subscribe(200, 550)
  548. );
  549. }
  550. }
  551. }