CollectTest.cs 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  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.Concurrency;
  8. using System.Reactive.Linq;
  9. using Microsoft.Reactive.Testing;
  10. using Xunit;
  11. namespace ReactiveTests.Tests
  12. {
  13. public class CollectTest : ReactiveTest
  14. {
  15. [Fact]
  16. public void Collect_ArgumentChecking()
  17. {
  18. var someObservable = Observable.Empty<int>();
  19. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(default(IObservable<int>), () => 0, (x, y) => x));
  20. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(someObservable, default(Func<int>), (x, y) => x));
  21. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(someObservable, () => 0, default));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(default(IObservable<int>), () => 0, (x, y) => x, x => x));
  23. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(someObservable, default(Func<int>), (x, y) => x, x => x));
  24. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(someObservable, () => 0, default, x => x));
  25. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Collect(someObservable, () => 0, (x, y) => x, default));
  26. }
  27. [Fact]
  28. public void Collect_Regular1()
  29. {
  30. var scheduler = new TestScheduler();
  31. var xs = scheduler.CreateHotObservable(
  32. OnNext(100, 1),
  33. OnNext(200, 2),
  34. OnNext(300, 3),
  35. OnNext(400, 4),
  36. OnNext(500, 5),
  37. OnNext(600, 6),
  38. OnNext(700, 7),
  39. OnNext(800, 8),
  40. OnCompleted<int>(900)
  41. );
  42. var ys = xs.Collect(() => 0, (x, y) => x + y);
  43. var e = default(IEnumerator<int>);
  44. var res = new List<int>();
  45. var log = new Action(() =>
  46. {
  47. Assert.True(e.MoveNext());
  48. res.Add(e.Current);
  49. });
  50. scheduler.ScheduleAbsolute(250, () => { e = ys.GetEnumerator(); });
  51. scheduler.ScheduleAbsolute(270, log);
  52. scheduler.ScheduleAbsolute(310, log);
  53. scheduler.ScheduleAbsolute(450, log);
  54. scheduler.ScheduleAbsolute(470, log);
  55. scheduler.ScheduleAbsolute(750, log);
  56. scheduler.ScheduleAbsolute(850, log);
  57. scheduler.ScheduleAbsolute(950, log);
  58. scheduler.ScheduleAbsolute(980, () => Assert.False(e.MoveNext()));
  59. scheduler.Start();
  60. xs.Subscriptions.AssertEqual(
  61. Subscribe(250, 900)
  62. );
  63. Assert.Equal(7, res.Count);
  64. Assert.Equal(res[0], new int[] { }.Sum());
  65. Assert.Equal(res[1], new int[] { 3 }.Sum());
  66. Assert.Equal(res[2], new int[] { 4 }.Sum());
  67. Assert.Equal(res[3], new int[] { }.Sum());
  68. Assert.Equal(res[4], new int[] { 5, 6, 7 }.Sum());
  69. Assert.Equal(res[5], new int[] { 8 }.Sum());
  70. Assert.Equal(res[6], new int[] { }.Sum());
  71. }
  72. [Fact]
  73. public void Collect_Regular2()
  74. {
  75. var scheduler = new TestScheduler();
  76. var xs = scheduler.CreateHotObservable(
  77. OnNext(100, 1),
  78. OnNext(200, 2),
  79. OnNext(300, 3),
  80. OnNext(400, 4),
  81. OnNext(500, 5),
  82. OnNext(600, 6),
  83. OnNext(700, 7),
  84. OnNext(800, 8),
  85. OnCompleted<int>(900)
  86. );
  87. var ys = xs.Collect(() => 0, (x, y) => x + y);
  88. var e = default(IEnumerator<int>);
  89. var res = new List<int>();
  90. var log = new Action(() =>
  91. {
  92. Assert.True(e.MoveNext());
  93. res.Add(e.Current);
  94. });
  95. scheduler.ScheduleAbsolute(250, () => { e = ys.GetEnumerator(); });
  96. scheduler.ScheduleAbsolute(550, log);
  97. scheduler.ScheduleAbsolute(950, log);
  98. scheduler.ScheduleAbsolute(980, () => Assert.False(e.MoveNext()));
  99. scheduler.Start();
  100. xs.Subscriptions.AssertEqual(
  101. Subscribe(250, 900)
  102. );
  103. Assert.Equal(2, res.Count);
  104. Assert.Equal(res[0], new int[] { 3, 4, 5 }.Sum());
  105. Assert.Equal(res[1], new int[] { 6, 7, 8 }.Sum());
  106. }
  107. [Fact]
  108. public void Collect_InitialCollectorThrows()
  109. {
  110. var scheduler = new TestScheduler();
  111. var xs = scheduler.CreateHotObservable(
  112. OnNext(100, 1),
  113. OnNext(200, 2),
  114. OnNext(300, 3),
  115. OnNext(400, 4),
  116. OnCompleted<int>(500)
  117. );
  118. var ex = new Exception();
  119. var ys = xs.Collect<int, int>(() => { throw ex; }, (x, y) => x + y);
  120. var ex_ = default(Exception);
  121. scheduler.ScheduleAbsolute(250, () =>
  122. {
  123. try
  124. {
  125. ys.GetEnumerator();
  126. }
  127. catch (Exception err)
  128. {
  129. ex_ = err;
  130. }
  131. });
  132. scheduler.Start();
  133. xs.Subscriptions.AssertEqual(
  134. );
  135. Assert.Same(ex_, ex);
  136. }
  137. [Fact]
  138. public void Collect_SecondCollectorThrows()
  139. {
  140. var scheduler = new TestScheduler();
  141. var xs = scheduler.CreateHotObservable(
  142. OnNext(100, 1),
  143. OnNext(200, 2),
  144. OnNext(300, 3),
  145. OnNext(400, 4),
  146. OnCompleted<int>(500)
  147. );
  148. var ex = new Exception();
  149. var n = 0;
  150. var ys = xs.Collect(() =>
  151. {
  152. if (n++ == 0)
  153. {
  154. return 0;
  155. }
  156. else
  157. {
  158. throw ex;
  159. }
  160. }, (x, y) => x + y);
  161. var e = default(IEnumerator<int>);
  162. var ex_ = default(Exception);
  163. scheduler.ScheduleAbsolute(250, () => e = ys.GetEnumerator());
  164. scheduler.ScheduleAbsolute(350, () =>
  165. {
  166. try
  167. {
  168. e.MoveNext();
  169. }
  170. catch (Exception err)
  171. {
  172. ex_ = err;
  173. }
  174. });
  175. scheduler.Start();
  176. xs.Subscriptions.AssertEqual(
  177. Subscribe(250, 350)
  178. );
  179. Assert.Same(ex_, ex);
  180. }
  181. [Fact]
  182. public void Collect_NewCollectorThrows()
  183. {
  184. var scheduler = new TestScheduler();
  185. var xs = scheduler.CreateHotObservable(
  186. OnNext(100, 1),
  187. OnNext(200, 2),
  188. OnNext(300, 3),
  189. OnNext(400, 4),
  190. OnCompleted<int>(500)
  191. );
  192. var ex = new Exception();
  193. var ys = xs.Collect(() => 0, (x, y) => x + y, x => { throw ex; });
  194. var e = default(IEnumerator<int>);
  195. var ex_ = default(Exception);
  196. scheduler.ScheduleAbsolute(250, () => e = ys.GetEnumerator());
  197. scheduler.ScheduleAbsolute(350, () =>
  198. {
  199. try
  200. {
  201. e.MoveNext();
  202. }
  203. catch (Exception err)
  204. {
  205. ex_ = err;
  206. }
  207. });
  208. scheduler.Start();
  209. xs.Subscriptions.AssertEqual(
  210. Subscribe(250, 350)
  211. );
  212. Assert.Same(ex_, ex);
  213. }
  214. [Fact]
  215. public void Collect_MergeThrows()
  216. {
  217. var scheduler = new TestScheduler();
  218. var xs = scheduler.CreateHotObservable(
  219. OnNext(100, 1),
  220. OnNext(200, 2),
  221. OnNext(300, 3),
  222. OnNext(400, 4),
  223. OnCompleted<int>(500)
  224. );
  225. var ex = new Exception();
  226. var ys = xs.Collect(() => 0, (x, y) => { throw ex; });
  227. var e = default(IEnumerator<int>);
  228. var ex_ = default(Exception);
  229. scheduler.ScheduleAbsolute(250, () => { e = ys.GetEnumerator(); });
  230. scheduler.ScheduleAbsolute(350, () =>
  231. {
  232. try
  233. {
  234. e.MoveNext();
  235. }
  236. catch (Exception err)
  237. {
  238. ex_ = err;
  239. }
  240. });
  241. scheduler.Start();
  242. xs.Subscriptions.AssertEqual(
  243. Subscribe(250, 300)
  244. );
  245. Assert.Same(ex_, ex);
  246. }
  247. }
  248. }