1
0

Observable.Imperative.cs 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
  2. using System.Collections.Generic;
  3. using System.Reactive.Concurrency;
  4. using System.Threading;
  5. #if !NO_TPL
  6. using System.Threading.Tasks;
  7. #endif
  8. namespace System.Reactive.Linq
  9. {
  10. public static partial class Observable
  11. {
  12. #region + ForEachAsync +
  13. #if !NO_TPL
  14. /// <summary>
  15. /// Invokes an action for each element in the observable sequence, and returns a Task object that will get signaled when the sequence terminates.
  16. /// </summary>
  17. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  18. /// <param name="source">Source sequence.</param>
  19. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  20. /// <returns>Task that signals the termination of the sequence.</returns>
  21. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  22. /// <remarks>This operator is especially useful in conjunction with the asynchronous programming features introduced in C# 5.0 and Visual Basic 11.</remarks>
  23. public static Task ForEachAsync<TSource>(this IObservable<TSource> source, Action<TSource> onNext)
  24. {
  25. if (source == null)
  26. throw new ArgumentNullException("source");
  27. if (onNext == null)
  28. throw new ArgumentNullException("onNext");
  29. return s_impl.ForEachAsync<TSource>(source, onNext);
  30. }
  31. /// <summary>
  32. /// Invokes an action for each element in the observable sequence, and returns a Task object that will get signaled when the sequence terminates.
  33. /// The loop can be quit prematurely by setting the specified cancellation token.
  34. /// </summary>
  35. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  36. /// <param name="source">Source sequence.</param>
  37. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  38. /// <param name="cancellationToken">Cancellation token used to stop the loop.</param>
  39. /// <returns>Task that signals the termination of the sequence.</returns>
  40. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  41. /// <remarks>This operator is especially useful in conjunction with the asynchronous programming features introduced in C# 5.0 and Visual Basic 11.</remarks>
  42. public static Task ForEachAsync<TSource>(this IObservable<TSource> source, Action<TSource> onNext, CancellationToken cancellationToken)
  43. {
  44. if (source == null)
  45. throw new ArgumentNullException("source");
  46. if (onNext == null)
  47. throw new ArgumentNullException("onNext");
  48. return s_impl.ForEachAsync<TSource>(source, onNext, cancellationToken);
  49. }
  50. /// <summary>
  51. /// Invokes an action for each element in the observable sequence, incorporating the element's index, and returns a Task object that will get signaled when the sequence terminates.
  52. /// </summary>
  53. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  54. /// <param name="source">Source sequence.</param>
  55. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  56. /// <returns>Task that signals the termination of the sequence.</returns>
  57. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  58. /// <remarks>This operator is especially useful in conjunction with the asynchronous programming features introduced in C# 5.0 and Visual Basic 11.</remarks>
  59. public static Task ForEachAsync<TSource>(this IObservable<TSource> source, Action<TSource, int> onNext)
  60. {
  61. if (source == null)
  62. throw new ArgumentNullException("source");
  63. if (onNext == null)
  64. throw new ArgumentNullException("onNext");
  65. return s_impl.ForEachAsync<TSource>(source, onNext);
  66. }
  67. /// <summary>
  68. /// Invokes an action for each element in the observable sequence, incorporating the element's index, and returns a Task object that will get signaled when the sequence terminates.
  69. /// The loop can be quit prematurely by setting the specified cancellation token.
  70. /// </summary>
  71. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  72. /// <param name="source">Source sequence.</param>
  73. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  74. /// <param name="cancellationToken">Cancellation token used to stop the loop.</param>
  75. /// <returns>Task that signals the termination of the sequence.</returns>
  76. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  77. /// <remarks>This operator is especially useful in conjunction with the asynchronous programming features introduced in C# 5.0 and Visual Basic 11.</remarks>
  78. public static Task ForEachAsync<TSource>(this IObservable<TSource> source, Action<TSource, int> onNext, CancellationToken cancellationToken)
  79. {
  80. if (source == null)
  81. throw new ArgumentNullException("source");
  82. if (onNext == null)
  83. throw new ArgumentNullException("onNext");
  84. return s_impl.ForEachAsync<TSource>(source, onNext, cancellationToken);
  85. }
  86. #endif
  87. #endregion
  88. #region + Case +
  89. /// <summary>
  90. /// Uses <paramref name="selector"/> to determine which source in <paramref name="sources"/> to return, choosing <paramref name="defaultSource"/> if no match is found.
  91. /// </summary>
  92. /// <typeparam name="TValue">The type of the value returned by the selector function, used to look up the resulting source.</typeparam>
  93. /// <typeparam name="TResult">The type of the elements in the result sequence.</typeparam>
  94. /// <param name="selector">Selector function invoked to determine the source to lookup in the <paramref name="sources"/> dictionary.</param>
  95. /// <param name="sources">Dictionary of sources to select from based on the <paramref name="selector"/> invocation result.</param>
  96. /// <param name="defaultSource">Default source to select in case no matching source in <paramref name="sources"/> is found.</param>
  97. /// <returns>The observable sequence retrieved from the <paramref name="sources"/> dictionary based on the <paramref name="selector"/> invocation result, or <paramref name="defaultSource"/> if no match is found.</returns>
  98. /// <exception cref="ArgumentNullException"><paramref name="selector"/> or <paramref name="sources"/> or <paramref name="defaultSource"/> is null.</exception>
  99. public static IObservable<TResult> Case<TValue, TResult>(Func<TValue> selector, IDictionary<TValue, IObservable<TResult>> sources, IObservable<TResult> defaultSource)
  100. {
  101. if (selector == null)
  102. throw new ArgumentNullException("selector");
  103. if (sources == null)
  104. throw new ArgumentNullException("sources");
  105. if (defaultSource == null)
  106. throw new ArgumentNullException("defaultSource");
  107. return s_impl.Case<TValue, TResult>(selector, sources, defaultSource);
  108. }
  109. /// <summary>
  110. /// Uses <paramref name="selector"/> to determine which source in <paramref name="sources"/> to return, choosing an empty sequence on the specified scheduler if no match is found.
  111. /// </summary>
  112. /// <typeparam name="TValue">The type of the value returned by the selector function, used to look up the resulting source.</typeparam>
  113. /// <typeparam name="TResult">The type of the elements in the result sequence.</typeparam>
  114. /// <param name="selector">Selector function invoked to determine the source to lookup in the <paramref name="sources"/> dictionary.</param>
  115. /// <param name="sources">Dictionary of sources to select from based on the <paramref name="selector"/> invocation result.</param>
  116. /// <param name="scheduler">Scheduler to generate an empty sequence on in case no matching source in <paramref name="sources"/> is found.</param>
  117. /// <returns>The observable sequence retrieved from the <paramref name="sources"/> dictionary based on the <paramref name="selector"/> invocation result, or an empty sequence if no match is found.</returns>
  118. /// <exception cref="ArgumentNullException"><paramref name="selector"/> or <paramref name="sources"/> or <paramref name="scheduler"/> is null.</exception>
  119. public static IObservable<TResult> Case<TValue, TResult>(Func<TValue> selector, IDictionary<TValue, IObservable<TResult>> sources, IScheduler scheduler)
  120. {
  121. if (selector == null)
  122. throw new ArgumentNullException("selector");
  123. if (sources == null)
  124. throw new ArgumentNullException("sources");
  125. if (scheduler == null)
  126. throw new ArgumentNullException("scheduler");
  127. return s_impl.Case<TValue, TResult>(selector, sources, scheduler);
  128. }
  129. /// <summary>
  130. /// Uses <paramref name="selector"/> to determine which source in <paramref name="sources"/> to return, choosing an empty sequence if no match is found.
  131. /// </summary>
  132. /// <typeparam name="TValue">The type of the value returned by the selector function, used to look up the resulting source.</typeparam>
  133. /// <typeparam name="TResult">The type of the elements in the result sequence.</typeparam>
  134. /// <param name="selector">Selector function invoked to determine the source to lookup in the <paramref name="sources"/> dictionary.</param>
  135. /// <param name="sources">Dictionary of sources to select from based on the <paramref name="selector"/> invocation result.</param>
  136. /// <returns>The observable sequence retrieved from the <paramref name="sources"/> dictionary based on the <paramref name="selector"/> invocation result, or an empty sequence if no match is found.</returns>
  137. /// <exception cref="ArgumentNullException"><paramref name="selector"/> or <paramref name="sources"/> is null.</exception>
  138. public static IObservable<TResult> Case<TValue, TResult>(Func<TValue> selector, IDictionary<TValue, IObservable<TResult>> sources)
  139. {
  140. if (selector == null)
  141. throw new ArgumentNullException("selector");
  142. if (sources == null)
  143. throw new ArgumentNullException("sources");
  144. return s_impl.Case<TValue, TResult>(selector, sources);
  145. }
  146. #endregion
  147. #region + DoWhile +
  148. /// <summary>
  149. /// Repeats the given <paramref name="source"/> as long as the specified <paramref name="condition"/> holds, where the <paramref name="condition"/> is evaluated after each repeated <paramref name="source"/> completed.
  150. /// </summary>
  151. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  152. /// <param name="source">Source to repeat as long as the <paramref name="condition"/> function evaluates to true.</param>
  153. /// <param name="condition">Condition that will be evaluated upon the completion of an iteration through the <paramref name="source"/>, to determine whether repetition of the source is required.</param>
  154. /// <returns>The observable sequence obtained by concatenating the <paramref name="source"/> sequence as long as the <paramref name="condition"/> holds.</returns>
  155. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="condition"/> is null.</exception>
  156. public static IObservable<TSource> DoWhile<TSource>(this IObservable<TSource> source, Func<bool> condition)
  157. {
  158. if (source == null)
  159. throw new ArgumentNullException("source");
  160. if (condition == null)
  161. throw new ArgumentNullException("condition");
  162. return s_impl.DoWhile<TSource>(source, condition);
  163. }
  164. #endregion
  165. #region + For +
  166. /// <summary>
  167. /// Concatenates the observable sequences obtained by running the <paramref name="resultSelector"/> for each element in the given enumerable <paramref name="source"/>.
  168. /// </summary>
  169. /// <typeparam name="TSource">The type of the elements in the enumerable source sequence.</typeparam>
  170. /// <typeparam name="TResult">The type of the elements in the observable result sequence.</typeparam>
  171. /// <param name="source">Enumerable source for which each element will be mapped onto an observable source that will be concatenated in the result sequence.</param>
  172. /// <param name="resultSelector">Function to select an observable source for each element in the <paramref name="source"/>.</param>
  173. /// <returns>The observable sequence obtained by concatenating the sources returned by <paramref name="resultSelector"/> for each element in the <paramref name="source"/>.</returns>
  174. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="resultSelector"/> is null.</exception>
  175. public static IObservable<TResult> For<TSource, TResult>(IEnumerable<TSource> source, Func<TSource, IObservable<TResult>> resultSelector)
  176. {
  177. if (source == null)
  178. throw new ArgumentNullException("source");
  179. if (resultSelector == null)
  180. throw new ArgumentNullException("resultSelector");
  181. return s_impl.For<TSource, TResult>(source, resultSelector);
  182. }
  183. #endregion
  184. #region + If +
  185. /// <summary>
  186. /// If the specified <paramref name="condition"/> evaluates true, select the <paramref name="thenSource"/> sequence. Otherwise, select the <paramref name="elseSource"/> sequence.
  187. /// </summary>
  188. /// <typeparam name="TResult">The type of the elements in the result sequence.</typeparam>
  189. /// <param name="condition">Condition evaluated to decide which sequence to return.</param>
  190. /// <param name="thenSource">Sequence returned in case <paramref name="condition"/> evaluates true.</param>
  191. /// <param name="elseSource">Sequence returned in case <paramref name="condition"/> evaluates false.</param>
  192. /// <returns><paramref name="thenSource"/> if <paramref name="condition"/> evaluates true; <paramref name="elseSource"/> otherwise.</returns>
  193. /// <exception cref="ArgumentNullException"><paramref name="condition"/> or <paramref name="thenSource"/> or <paramref name="elseSource"/> is null.</exception>
  194. public static IObservable<TResult> If<TResult>(Func<bool> condition, IObservable<TResult> thenSource, IObservable<TResult> elseSource)
  195. {
  196. if (condition == null)
  197. throw new ArgumentNullException("condition");
  198. if (thenSource == null)
  199. throw new ArgumentNullException("thenSource");
  200. if (elseSource == null)
  201. throw new ArgumentNullException("elseSource");
  202. return s_impl.If<TResult>(condition, thenSource, elseSource);
  203. }
  204. /// <summary>
  205. /// If the specified <paramref name="condition"/> evaluates true, select the <paramref name="thenSource"/> sequence. Otherwise, return an empty sequence.
  206. /// </summary>
  207. /// <typeparam name="TResult">The type of the elements in the result sequence.</typeparam>
  208. /// <param name="condition">Condition evaluated to decide which sequence to return.</param>
  209. /// <param name="thenSource">Sequence returned in case <paramref name="condition"/> evaluates true.</param>
  210. /// <returns><paramref name="thenSource"/> if <paramref name="condition"/> evaluates true; an empty sequence otherwise.</returns>
  211. /// <exception cref="ArgumentNullException"><paramref name="condition"/> or <paramref name="thenSource"/> is null.</exception>
  212. public static IObservable<TResult> If<TResult>(Func<bool> condition, IObservable<TResult> thenSource)
  213. {
  214. if (condition == null)
  215. throw new ArgumentNullException("condition");
  216. if (thenSource == null)
  217. throw new ArgumentNullException("thenSource");
  218. return s_impl.If<TResult>(condition, thenSource);
  219. }
  220. /// <summary>
  221. /// If the specified <paramref name="condition"/> evaluates true, select the <paramref name="thenSource"/> sequence. Otherwise, return an empty sequence generated on the specified scheduler.
  222. /// </summary>
  223. /// <typeparam name="TResult">The type of the elements in the result sequence.</typeparam>
  224. /// <param name="condition">Condition evaluated to decide which sequence to return.</param>
  225. /// <param name="thenSource">Sequence returned in case <paramref name="condition"/> evaluates true.</param>
  226. /// <param name="scheduler">Scheduler to generate an empty sequence on in case <paramref name="condition"/> evaluates false.</param>
  227. /// <returns><paramref name="thenSource"/> if <paramref name="condition"/> evaluates true; an empty sequence otherwise.</returns>
  228. /// <exception cref="ArgumentNullException"><paramref name="condition"/> or <paramref name="thenSource"/> or <paramref name="scheduler"/> is null.</exception>
  229. public static IObservable<TResult> If<TResult>(Func<bool> condition, IObservable<TResult> thenSource, IScheduler scheduler)
  230. {
  231. if (condition == null)
  232. throw new ArgumentNullException("condition");
  233. if (thenSource == null)
  234. throw new ArgumentNullException("thenSource");
  235. if (scheduler == null)
  236. throw new ArgumentNullException("scheduler");
  237. return s_impl.If<TResult>(condition, thenSource, scheduler);
  238. }
  239. #endregion
  240. #region + While +
  241. /// <summary>
  242. /// Repeats the given <paramref name="source"/> as long as the specified <paramref name="condition"/> holds, where the <paramref name="condition"/> is evaluated before each repeated <paramref name="source"/> is subscribed to.
  243. /// </summary>
  244. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  245. /// <param name="source">Source to repeat as long as the <paramref name="condition"/> function evaluates to true.</param>
  246. /// <param name="condition">Condition that will be evaluated before subscription to the <paramref name="source"/>, to determine whether repetition of the source is required.</param>
  247. /// <returns>The observable sequence obtained by concatenating the <paramref name="source"/> sequence as long as the <paramref name="condition"/> holds.</returns>
  248. /// <exception cref="ArgumentNullException"><paramref name="condition"/> or <paramref name="source"/> is null.</exception>
  249. public static IObservable<TSource> While<TSource>(Func<bool> condition, IObservable<TSource> source)
  250. {
  251. if (condition == null)
  252. throw new ArgumentNullException("condition");
  253. if (source == null)
  254. throw new ArgumentNullException("source");
  255. return s_impl.While<TSource>(condition, source);
  256. }
  257. #endregion
  258. }
  259. }