ToObservableTest.cs 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220
  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 System.Runtime.CompilerServices;
  10. using System.Threading;
  11. using Microsoft.Reactive.Testing;
  12. using ReactiveTests.Dummies;
  13. using Xunit;
  14. namespace ReactiveTests.Tests
  15. {
  16. public class ToObservableTest : ReactiveTest
  17. {
  18. [Fact]
  19. public void EnumerableToObservable_ArgumentChecking()
  20. {
  21. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.ToObservable((IEnumerable<int>)null, DummyScheduler.Instance));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.ToObservable(DummyEnumerable<int>.Instance, null));
  23. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.ToObservable(DummyEnumerable<int>.Instance, DummyScheduler.Instance).Subscribe(null));
  24. ReactiveAssert.Throws<NullReferenceException>(() => Observable.ToObservable(NullEnumeratorEnumerable<int>.Instance, Scheduler.CurrentThread).Subscribe());
  25. }
  26. [Fact]
  27. public void EnumerableToObservable_Complete()
  28. {
  29. var scheduler = new TestScheduler();
  30. var e = new MockEnumerable<int>(scheduler,
  31. new[] { 3, 1, 2, 4 }
  32. );
  33. var results = scheduler.Start(() =>
  34. e.ToObservable(scheduler)
  35. );
  36. results.Messages.AssertEqual(
  37. OnNext(201, 3),
  38. OnNext(202, 1),
  39. OnNext(203, 2),
  40. OnNext(204, 4),
  41. OnCompleted<int>(205)
  42. );
  43. e.Subscriptions.AssertEqual(
  44. Subscribe(200, 205)
  45. );
  46. }
  47. [Fact]
  48. public void EnumerableToObservable_Dispose()
  49. {
  50. var scheduler = new TestScheduler();
  51. var e = new MockEnumerable<int>(scheduler,
  52. new[] { 3, 1, 2, 4 }
  53. );
  54. var results = scheduler.Start(() =>
  55. e.ToObservable(scheduler),
  56. 203
  57. );
  58. results.Messages.AssertEqual(
  59. OnNext(201, 3),
  60. OnNext(202, 1)
  61. );
  62. e.Subscriptions.AssertEqual(
  63. Subscribe(200, 203)
  64. );
  65. }
  66. [Fact]
  67. public void EnumerableToObservable_Error()
  68. {
  69. var scheduler = new TestScheduler();
  70. var ex = new Exception();
  71. var e = new MockEnumerable<int>(scheduler,
  72. EnumerableToObservable_Error_Core(ex)
  73. );
  74. var results = scheduler.Start(() =>
  75. e.ToObservable(scheduler)
  76. );
  77. results.Messages.AssertEqual(
  78. OnNext(201, 1),
  79. OnNext(202, 2),
  80. OnError<int>(203, ex)
  81. );
  82. e.Subscriptions.AssertEqual(
  83. Subscribe(200, 203)
  84. );
  85. }
  86. [Fact]
  87. public void EnumerableToObservable_Default_ArgumentChecking()
  88. {
  89. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.ToObservable((IEnumerable<int>)null));
  90. ReactiveAssert.Throws<ArgumentNullException>(() => Observable.ToObservable(DummyEnumerable<int>.Instance).Subscribe(null));
  91. }
  92. [Fact]
  93. public void EnumerableToObservable_Default()
  94. {
  95. var xs = new[] { 4, 3, 1, 5, 9, 2 };
  96. xs.ToObservable().AssertEqual(xs.ToObservable(DefaultScheduler.Instance));
  97. }
  98. #if !NO_PERF
  99. [Fact]
  100. public void EnumerableToObservable_LongRunning_Complete()
  101. {
  102. var start = default(ManualResetEvent);
  103. var end = default(ManualResetEvent);
  104. var scheduler = new TestLongRunningScheduler(x => start = x, x => end = x);
  105. var e = new[] { 3, 1, 2, 4 };
  106. var results = e.ToObservable(scheduler);
  107. var lst = new List<int>();
  108. results.Subscribe(lst.Add);
  109. start.WaitOne();
  110. end.WaitOne();
  111. Assert.True(e.SequenceEqual(lst));
  112. }
  113. [Fact]
  114. [MethodImpl(MethodImplOptions.NoOptimization)]
  115. public void EnumerableToObservable_LongRunning_Dispose()
  116. {
  117. var start = default(ManualResetEvent);
  118. var end = default(ManualResetEvent);
  119. var scheduler = new TestLongRunningScheduler(x => start = x, x => end = x);
  120. var e = Enumerable.Range(0, int.MaxValue);
  121. var results = e.ToObservable(scheduler);
  122. var lst = new List<int>();
  123. var d = results.Subscribe(lst.Add);
  124. start.WaitOne();
  125. while (lst.Count < 100)
  126. {
  127. ;
  128. }
  129. d.Dispose();
  130. end.WaitOne();
  131. Assert.True(e.Take(100).SequenceEqual(lst.Take(100)));
  132. }
  133. [Fact]
  134. public void EnumerableToObservable_LongRunning_Error()
  135. {
  136. var start = default(ManualResetEvent);
  137. var end = default(ManualResetEvent);
  138. var scheduler = new TestLongRunningScheduler(x => start = x, x => end = x);
  139. var ex = new Exception();
  140. var e = EnumerableToObservable_Error_Core(ex);
  141. var results = e.ToObservable(scheduler);
  142. var lst = new List<int>();
  143. var err = default(Exception);
  144. results.Subscribe(lst.Add, ex_ => err = ex_);
  145. start.WaitOne();
  146. end.WaitOne();
  147. Assert.True(new[] { 1, 2 }.SequenceEqual(lst));
  148. Assert.Same(ex, err);
  149. }
  150. #endif
  151. private static IEnumerable<int> EnumerableToObservable_Error_Core(Exception ex)
  152. {
  153. yield return 1;
  154. yield return 2;
  155. throw ex;
  156. }
  157. [Fact]
  158. public void EnumerableToObservable_GetEnumeratorThrows()
  159. {
  160. var ex = new Exception();
  161. var scheduler = new TestScheduler();
  162. var xs = new RogueEnumerable<int>(ex);
  163. var res = scheduler.Start(() =>
  164. xs.ToObservable(scheduler)
  165. );
  166. res.Messages.AssertEqual(
  167. OnError<int>(200, ex)
  168. );
  169. }
  170. }
  171. }