MulticastTest.cs 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483
  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.Text;
  8. using System.Threading.Tasks;
  9. using System.Reactive;
  10. using System.Reactive.Concurrency;
  11. using System.Reactive.Linq;
  12. using Microsoft.Reactive.Testing;
  13. using Xunit;
  14. using ReactiveTests.Dummies;
  15. using System.Reflection;
  16. using System.Reactive.Subjects;
  17. namespace ReactiveTests.Tests
  18. {
  19. public class MulticastTest : ReactiveTest
  20. {
  21. [Fact]
  22. public void Multicast_ArgumentChecking()
  23. {
  24. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Multicast<int, int>(null, new Subject<int>()));
  25. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Multicast<int, int>(DummyObservable<int>.Instance, null));
  26. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Multicast<int, int, int>(null, () => new Subject<int>(), xs => xs));
  27. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Multicast<int, int, int>(DummyObservable<int>.Instance, null, xs => xs));
  28. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.Multicast<int, int, int>(DummyObservable<int>.Instance, () => new Subject<int>(), null));
  29. }
  30. [Fact]
  31. public void Multicast_Hot_1()
  32. {
  33. var scheduler = new TestScheduler();
  34. var s = new Subject<int>();
  35. var xs = scheduler.CreateHotObservable(
  36. OnNext(40, 0),
  37. OnNext(90, 1),
  38. OnNext(150, 2),
  39. OnNext(210, 3),
  40. OnNext(240, 4),
  41. OnNext(270, 5),
  42. OnNext(330, 6),
  43. OnNext(340, 7),
  44. OnCompleted<int>(390)
  45. );
  46. var c = default(IConnectableObservable<int>);
  47. var o = scheduler.CreateObserver<int>();
  48. var d1 = default(IDisposable);
  49. var d2 = default(IDisposable);
  50. scheduler.ScheduleAbsolute(50, () => c = xs.Multicast(s));
  51. scheduler.ScheduleAbsolute(100, () => d1 = c.Subscribe(o));
  52. scheduler.ScheduleAbsolute(200, () => d2 = c.Connect());
  53. scheduler.ScheduleAbsolute(300, () => d1.Dispose());
  54. scheduler.Start();
  55. o.Messages.AssertEqual(
  56. OnNext(210, 3),
  57. OnNext(240, 4),
  58. OnNext(270, 5)
  59. );
  60. xs.Subscriptions.AssertEqual(
  61. Subscribe(200, 390)
  62. );
  63. }
  64. [Fact]
  65. public void Multicast_Hot_2()
  66. {
  67. var scheduler = new TestScheduler();
  68. var s = new Subject<int>();
  69. var xs = scheduler.CreateHotObservable(
  70. OnNext(40, 0),
  71. OnNext(90, 1),
  72. OnNext(150, 2),
  73. OnNext(210, 3),
  74. OnNext(240, 4),
  75. OnNext(270, 5),
  76. OnNext(330, 6),
  77. OnNext(340, 7),
  78. OnCompleted<int>(390)
  79. );
  80. var c = default(IConnectableObservable<int>);
  81. var o = scheduler.CreateObserver<int>();
  82. var d1 = default(IDisposable);
  83. var d2 = default(IDisposable);
  84. scheduler.ScheduleAbsolute(50, () => c = xs.Multicast(s));
  85. scheduler.ScheduleAbsolute(100, () => d2 = c.Connect());
  86. scheduler.ScheduleAbsolute(200, () => d1 = c.Subscribe(o));
  87. scheduler.ScheduleAbsolute(300, () => d1.Dispose());
  88. scheduler.Start();
  89. o.Messages.AssertEqual(
  90. OnNext(210, 3),
  91. OnNext(240, 4),
  92. OnNext(270, 5)
  93. );
  94. xs.Subscriptions.AssertEqual(
  95. Subscribe(100, 390)
  96. );
  97. }
  98. [Fact]
  99. public void Multicast_Hot_3()
  100. {
  101. var scheduler = new TestScheduler();
  102. var s = new Subject<int>();
  103. var xs = scheduler.CreateHotObservable(
  104. OnNext(40, 0),
  105. OnNext(90, 1),
  106. OnNext(150, 2),
  107. OnNext(210, 3),
  108. OnNext(240, 4),
  109. OnNext(270, 5),
  110. OnNext(330, 6),
  111. OnNext(340, 7),
  112. OnCompleted<int>(390)
  113. );
  114. var c = default(IConnectableObservable<int>);
  115. var o = scheduler.CreateObserver<int>();
  116. var d1 = default(IDisposable);
  117. var d2 = default(IDisposable);
  118. scheduler.ScheduleAbsolute(50, () => c = xs.Multicast(s));
  119. scheduler.ScheduleAbsolute(100, () => d2 = c.Connect());
  120. scheduler.ScheduleAbsolute(200, () => d1 = c.Subscribe(o));
  121. scheduler.ScheduleAbsolute(300, () => d2.Dispose());
  122. scheduler.ScheduleAbsolute(335, () => d2 = c.Connect());
  123. scheduler.Start();
  124. o.Messages.AssertEqual(
  125. OnNext(210, 3),
  126. OnNext(240, 4),
  127. OnNext(270, 5),
  128. OnNext(340, 7),
  129. OnCompleted<int>(390)
  130. );
  131. xs.Subscriptions.AssertEqual(
  132. Subscribe(100, 300),
  133. Subscribe(335, 390)
  134. );
  135. }
  136. [Fact]
  137. public void Multicast_Hot_4()
  138. {
  139. var scheduler = new TestScheduler();
  140. var s = new Subject<int>();
  141. var ex = new Exception();
  142. var xs = scheduler.CreateHotObservable(
  143. OnNext(40, 0),
  144. OnNext(90, 1),
  145. OnNext(150, 2),
  146. OnNext(210, 3),
  147. OnNext(240, 4),
  148. OnNext(270, 5),
  149. OnNext(330, 6),
  150. OnNext(340, 7),
  151. OnError<int>(390, ex)
  152. );
  153. var c = default(IConnectableObservable<int>);
  154. var o = scheduler.CreateObserver<int>();
  155. var d1 = default(IDisposable);
  156. var d2 = default(IDisposable);
  157. scheduler.ScheduleAbsolute(50, () => c = xs.Multicast(s));
  158. scheduler.ScheduleAbsolute(100, () => d2 = c.Connect());
  159. scheduler.ScheduleAbsolute(200, () => d1 = c.Subscribe(o));
  160. scheduler.ScheduleAbsolute(300, () => d2.Dispose());
  161. scheduler.ScheduleAbsolute(335, () => d2 = c.Connect());
  162. scheduler.Start();
  163. o.Messages.AssertEqual(
  164. OnNext(210, 3),
  165. OnNext(240, 4),
  166. OnNext(270, 5),
  167. OnNext(340, 7),
  168. OnError<int>(390, ex)
  169. );
  170. xs.Subscriptions.AssertEqual(
  171. Subscribe(100, 300),
  172. Subscribe(335, 390)
  173. );
  174. }
  175. [Fact]
  176. public void Multicast_Hot_5()
  177. {
  178. var scheduler = new TestScheduler();
  179. var s = new Subject<int>();
  180. var ex = new Exception();
  181. var xs = scheduler.CreateHotObservable(
  182. OnNext(40, 0),
  183. OnNext(90, 1),
  184. OnNext(150, 2),
  185. OnNext(210, 3),
  186. OnNext(240, 4),
  187. OnNext(270, 5),
  188. OnNext(330, 6),
  189. OnNext(340, 7),
  190. OnError<int>(390, ex)
  191. );
  192. var c = default(IConnectableObservable<int>);
  193. var o = scheduler.CreateObserver<int>();
  194. var d1 = default(IDisposable);
  195. var d2 = default(IDisposable);
  196. scheduler.ScheduleAbsolute(50, () => c = xs.Multicast(s));
  197. scheduler.ScheduleAbsolute(100, () => d2 = c.Connect());
  198. scheduler.ScheduleAbsolute(400, () => d1 = c.Subscribe(o));
  199. scheduler.Start();
  200. o.Messages.AssertEqual(
  201. OnError<int>(400, ex)
  202. );
  203. xs.Subscriptions.AssertEqual(
  204. Subscribe(100, 390)
  205. );
  206. }
  207. [Fact]
  208. public void Multicast_Hot_6()
  209. {
  210. var scheduler = new TestScheduler();
  211. var s = new Subject<int>();
  212. var xs = scheduler.CreateHotObservable(
  213. OnNext(40, 0),
  214. OnNext(90, 1),
  215. OnNext(150, 2),
  216. OnNext(210, 3),
  217. OnNext(240, 4),
  218. OnNext(270, 5),
  219. OnNext(330, 6),
  220. OnNext(340, 7),
  221. OnCompleted<int>(390)
  222. );
  223. var c = default(IConnectableObservable<int>);
  224. var o = scheduler.CreateObserver<int>();
  225. var d1 = default(IDisposable);
  226. var d2 = default(IDisposable);
  227. scheduler.ScheduleAbsolute(50, () => c = xs.Multicast(s));
  228. scheduler.ScheduleAbsolute(100, () => d2 = c.Connect());
  229. scheduler.ScheduleAbsolute(400, () => d1 = c.Subscribe(o));
  230. scheduler.Start();
  231. o.Messages.AssertEqual(
  232. OnCompleted<int>(400)
  233. );
  234. xs.Subscriptions.AssertEqual(
  235. Subscribe(100, 390)
  236. );
  237. }
  238. [Fact]
  239. public void Multicast_Cold_Completed()
  240. {
  241. var scheduler = new TestScheduler();
  242. var xs = scheduler.CreateHotObservable(
  243. OnNext(40, 0),
  244. OnNext(90, 1),
  245. OnNext(150, 2),
  246. OnNext(210, 3),
  247. OnNext(240, 4),
  248. OnNext(270, 5),
  249. OnNext(330, 6),
  250. OnNext(340, 7),
  251. OnCompleted<int>(390)
  252. );
  253. var res = scheduler.Start(() =>
  254. xs.Multicast(() => new Subject<int>(), ys => ys)
  255. );
  256. res.Messages.AssertEqual(
  257. OnNext(210, 3),
  258. OnNext(240, 4),
  259. OnNext(270, 5),
  260. OnNext(330, 6),
  261. OnNext(340, 7),
  262. OnCompleted<int>(390)
  263. );
  264. xs.Subscriptions.AssertEqual(
  265. Subscribe(200, 390)
  266. );
  267. }
  268. [Fact]
  269. public void Multicast_Cold_Error()
  270. {
  271. var scheduler = new TestScheduler();
  272. var ex = new Exception();
  273. var xs = scheduler.CreateHotObservable(
  274. OnNext(40, 0),
  275. OnNext(90, 1),
  276. OnNext(150, 2),
  277. OnNext(210, 3),
  278. OnNext(240, 4),
  279. OnNext(270, 5),
  280. OnNext(330, 6),
  281. OnNext(340, 7),
  282. OnError<int>(390, ex)
  283. );
  284. var res = scheduler.Start(() =>
  285. xs.Multicast(() => new Subject<int>(), ys => ys)
  286. );
  287. res.Messages.AssertEqual(
  288. OnNext(210, 3),
  289. OnNext(240, 4),
  290. OnNext(270, 5),
  291. OnNext(330, 6),
  292. OnNext(340, 7),
  293. OnError<int>(390, ex)
  294. );
  295. xs.Subscriptions.AssertEqual(
  296. Subscribe(200, 390)
  297. );
  298. }
  299. [Fact]
  300. public void Multicast_Cold_Dispose()
  301. {
  302. var scheduler = new TestScheduler();
  303. var xs = scheduler.CreateHotObservable(
  304. OnNext(40, 0),
  305. OnNext(90, 1),
  306. OnNext(150, 2),
  307. OnNext(210, 3),
  308. OnNext(240, 4),
  309. OnNext(270, 5),
  310. OnNext(330, 6),
  311. OnNext(340, 7)
  312. );
  313. var res = scheduler.Start(() =>
  314. xs.Multicast(() => new Subject<int>(), ys => ys)
  315. );
  316. res.Messages.AssertEqual(
  317. OnNext(210, 3),
  318. OnNext(240, 4),
  319. OnNext(270, 5),
  320. OnNext(330, 6),
  321. OnNext(340, 7)
  322. );
  323. xs.Subscriptions.AssertEqual(
  324. Subscribe(200, 1000)
  325. );
  326. }
  327. [Fact]
  328. public void Multicast_Cold_Zip()
  329. {
  330. var scheduler = new TestScheduler();
  331. var xs = scheduler.CreateHotObservable(
  332. OnNext(40, 0),
  333. OnNext(90, 1),
  334. OnNext(150, 2),
  335. OnNext(210, 3),
  336. OnNext(240, 4),
  337. OnNext(270, 5),
  338. OnNext(330, 6),
  339. OnNext(340, 7),
  340. OnCompleted<int>(390)
  341. );
  342. var res = scheduler.Start(() =>
  343. xs.Multicast(() => new Subject<int>(), ys => ys.Zip(ys, (a, b) => a + b))
  344. );
  345. res.Messages.AssertEqual(
  346. OnNext(210, 6),
  347. OnNext(240, 8),
  348. OnNext(270, 10),
  349. OnNext(330, 12),
  350. OnNext(340, 14),
  351. OnCompleted<int>(390)
  352. );
  353. xs.Subscriptions.AssertEqual(
  354. Subscribe(200, 390)
  355. );
  356. }
  357. [Fact]
  358. public void Multicast_SubjectSelectorThrows()
  359. {
  360. var ex = new Exception();
  361. var scheduler = new TestScheduler();
  362. var xs = scheduler.CreateHotObservable(
  363. OnNext(210, 1),
  364. OnNext(240, 2),
  365. OnCompleted<int>(300)
  366. );
  367. var res = scheduler.Start(() =>
  368. xs.Multicast<int, int, int>(() => { throw ex; }, _ => _)
  369. );
  370. res.Messages.AssertEqual(
  371. OnError<int>(200, ex)
  372. );
  373. xs.Subscriptions.AssertEqual(
  374. );
  375. }
  376. [Fact]
  377. public void Multicast_SelectorThrows()
  378. {
  379. var ex = new Exception();
  380. var scheduler = new TestScheduler();
  381. var xs = scheduler.CreateHotObservable(
  382. OnNext(210, 1),
  383. OnNext(240, 2),
  384. OnCompleted<int>(300)
  385. );
  386. var res = scheduler.Start(() =>
  387. xs.Multicast<int, int, int>(() => new Subject<int>(), _ => { throw ex; })
  388. );
  389. res.Messages.AssertEqual(
  390. OnError<int>(200, ex)
  391. );
  392. xs.Subscriptions.AssertEqual(
  393. );
  394. }
  395. }
  396. }