Observable.Extensions.cs 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364
  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.ComponentModel;
  5. using System.Reactive;
  6. using System.Reactive.Disposables;
  7. using System.Threading;
  8. namespace System
  9. {
  10. /// <summary>
  11. /// Provides a set of static methods for subscribing delegates to observables.
  12. /// </summary>
  13. public static class ObservableExtensions
  14. {
  15. #region Subscribe delegate-based overloads
  16. /// <summary>
  17. /// Subscribes to the observable sequence without specifying any handlers.
  18. /// This method can be used to evaluate the observable sequence for its side-effects only.
  19. /// </summary>
  20. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  21. /// <param name="source">Observable sequence to subscribe to.</param>
  22. /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
  23. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  24. public static IDisposable Subscribe<T>(this IObservable<T> source)
  25. {
  26. if (source == null)
  27. throw new ArgumentNullException("source");
  28. //
  29. // [OK] Use of unsafe Subscribe: non-pretentious constructor for an observer; this overload is not to be used internally.
  30. //
  31. return source.Subscribe/*Unsafe*/(new AnonymousObserver<T>(Stubs<T>.Ignore, Stubs.Throw, Stubs.Nop));
  32. }
  33. /// <summary>
  34. /// Subscribes an element handler to an observable sequence.
  35. /// </summary>
  36. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  37. /// <param name="source">Observable sequence to subscribe to.</param>
  38. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  39. /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
  40. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  41. public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext)
  42. {
  43. if (source == null)
  44. throw new ArgumentNullException("source");
  45. if (onNext == null)
  46. throw new ArgumentNullException("onNext");
  47. //
  48. // [OK] Use of unsafe Subscribe: non-pretentious constructor for an observer; this overload is not to be used internally.
  49. //
  50. return source.Subscribe/*Unsafe*/(new AnonymousObserver<T>(onNext, Stubs.Throw, Stubs.Nop));
  51. }
  52. /// <summary>
  53. /// Subscribes an element handler and an exception handler to an observable sequence.
  54. /// </summary>
  55. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  56. /// <param name="source">Observable sequence to subscribe to.</param>
  57. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  58. /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
  59. /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
  60. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onError"/> is null.</exception>
  61. public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError)
  62. {
  63. if (source == null)
  64. throw new ArgumentNullException("source");
  65. if (onNext == null)
  66. throw new ArgumentNullException("onNext");
  67. if (onError == null)
  68. throw new ArgumentNullException("onError");
  69. //
  70. // [OK] Use of unsafe Subscribe: non-pretentious constructor for an observer; this overload is not to be used internally.
  71. //
  72. return source.Subscribe/*Unsafe*/(new AnonymousObserver<T>(onNext, onError, Stubs.Nop));
  73. }
  74. /// <summary>
  75. /// Subscribes an element handler and a completion handler to an observable sequence.
  76. /// </summary>
  77. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  78. /// <param name="source">Observable sequence to subscribe to.</param>
  79. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  80. /// <param name="onCompleted">Action to invoke upon graceful termination of the observable sequence.</param>
  81. /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
  82. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onCompleted"/> is null.</exception>
  83. public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action onCompleted)
  84. {
  85. if (source == null)
  86. throw new ArgumentNullException("source");
  87. if (onNext == null)
  88. throw new ArgumentNullException("onNext");
  89. if (onCompleted == null)
  90. throw new ArgumentNullException("onCompleted");
  91. //
  92. // [OK] Use of unsafe Subscribe: non-pretentious constructor for an observer; this overload is not to be used internally.
  93. //
  94. return source.Subscribe/*Unsafe*/(new AnonymousObserver<T>(onNext, Stubs.Throw, onCompleted));
  95. }
  96. /// <summary>
  97. /// Subscribes an element handler, an exception handler, and a completion handler to an observable sequence.
  98. /// </summary>
  99. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  100. /// <param name="source">Observable sequence to subscribe to.</param>
  101. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  102. /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
  103. /// <param name="onCompleted">Action to invoke upon graceful termination of the observable sequence.</param>
  104. /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
  105. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onError"/> or <paramref name="onCompleted"/> is null.</exception>
  106. public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError, Action onCompleted)
  107. {
  108. if (source == null)
  109. throw new ArgumentNullException("source");
  110. if (onNext == null)
  111. throw new ArgumentNullException("onNext");
  112. if (onError == null)
  113. throw new ArgumentNullException("onError");
  114. if (onCompleted == null)
  115. throw new ArgumentNullException("onCompleted");
  116. //
  117. // [OK] Use of unsafe Subscribe: non-pretentious constructor for an observer; this overload is not to be used internally.
  118. //
  119. return source.Subscribe/*Unsafe*/(new AnonymousObserver<T>(onNext, onError, onCompleted));
  120. }
  121. #endregion
  122. #region Subscribe overloads with CancellationToken
  123. #if !NO_TPL
  124. /// <summary>
  125. /// Subscribes an observer to an observable sequence, using a CancellationToken to support unsubscription.
  126. /// </summary>
  127. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  128. /// <param name="source">Observable sequence to subscribe to.</param>
  129. /// <param name="observer">Observer to subscribe to the sequence.</param>
  130. /// <param name="token">CancellationToken that can be signaled to unsubscribe from the source sequence.</param>
  131. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="observer"/> is null.</exception>
  132. public static void Subscribe<T>(this IObservable<T> source, IObserver<T> observer, CancellationToken token)
  133. {
  134. if (source == null)
  135. throw new ArgumentNullException("source");
  136. if (observer == null)
  137. throw new ArgumentNullException("observer");
  138. source.Subscribe_(observer, token);
  139. }
  140. /// <summary>
  141. /// Subscribes to the observable sequence without specifying any handlers, using a CancellationToken to support unsubscription.
  142. /// This method can be used to evaluate the observable sequence for its side-effects only.
  143. /// </summary>
  144. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  145. /// <param name="source">Observable sequence to subscribe to.</param>
  146. /// <param name="token">CancellationToken that can be signaled to unsubscribe from the source sequence.</param>
  147. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  148. public static void Subscribe<T>(this IObservable<T> source, CancellationToken token)
  149. {
  150. if (source == null)
  151. throw new ArgumentNullException("source");
  152. source.Subscribe_(new AnonymousObserver<T>(Stubs<T>.Ignore, Stubs.Throw, Stubs.Nop), token);
  153. }
  154. /// <summary>
  155. /// Subscribes an element handler to an observable sequence, using a CancellationToken to support unsubscription.
  156. /// </summary>
  157. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  158. /// <param name="source">Observable sequence to subscribe to.</param>
  159. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  160. /// <param name="token">CancellationToken that can be signaled to unsubscribe from the source sequence.</param>
  161. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  162. public static void Subscribe<T>(this IObservable<T> source, Action<T> onNext, CancellationToken token)
  163. {
  164. if (source == null)
  165. throw new ArgumentNullException("source");
  166. if (onNext == null)
  167. throw new ArgumentNullException("onNext");
  168. source.Subscribe_(new AnonymousObserver<T>(onNext, Stubs.Throw, Stubs.Nop), token);
  169. }
  170. /// <summary>
  171. /// Subscribes an element handler and an exception handler to an observable sequence, using a CancellationToken to support unsubscription.
  172. /// </summary>
  173. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  174. /// <param name="source">Observable sequence to subscribe to.</param>
  175. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  176. /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
  177. /// <param name="token">CancellationToken that can be signaled to unsubscribe from the source sequence.</param>
  178. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onError"/> is null.</exception>
  179. public static void Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError, CancellationToken token)
  180. {
  181. if (source == null)
  182. throw new ArgumentNullException("source");
  183. if (onNext == null)
  184. throw new ArgumentNullException("onNext");
  185. if (onError == null)
  186. throw new ArgumentNullException("onError");
  187. source.Subscribe_(new AnonymousObserver<T>(onNext, onError, Stubs.Nop), token);
  188. }
  189. /// <summary>
  190. /// Subscribes an element handler and a completion handler to an observable sequence, using a CancellationToken to support unsubscription.
  191. /// </summary>
  192. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  193. /// <param name="source">Observable sequence to subscribe to.</param>
  194. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  195. /// <param name="onCompleted">Action to invoke upon graceful termination of the observable sequence.</param>
  196. /// <param name="token">CancellationToken that can be signaled to unsubscribe from the source sequence.</param>
  197. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onCompleted"/> is null.</exception>
  198. public static void Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action onCompleted, CancellationToken token)
  199. {
  200. if (source == null)
  201. throw new ArgumentNullException("source");
  202. if (onNext == null)
  203. throw new ArgumentNullException("onNext");
  204. if (onCompleted == null)
  205. throw new ArgumentNullException("onCompleted");
  206. source.Subscribe_(new AnonymousObserver<T>(onNext, Stubs.Throw, onCompleted), token);
  207. }
  208. /// <summary>
  209. /// Subscribes an element handler, an exception handler, and a completion handler to an observable sequence, using a CancellationToken to support unsubscription.
  210. /// </summary>
  211. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  212. /// <param name="source">Observable sequence to subscribe to.</param>
  213. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  214. /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
  215. /// <param name="onCompleted">Action to invoke upon graceful termination of the observable sequence.</param>
  216. /// <param name="token">CancellationToken that can be signaled to unsubscribe from the source sequence.</param>
  217. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onError"/> or <paramref name="onCompleted"/> is null.</exception>
  218. public static void Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError, Action onCompleted, CancellationToken token)
  219. {
  220. if (source == null)
  221. throw new ArgumentNullException("source");
  222. if (onNext == null)
  223. throw new ArgumentNullException("onNext");
  224. if (onError == null)
  225. throw new ArgumentNullException("onError");
  226. if (onCompleted == null)
  227. throw new ArgumentNullException("onCompleted");
  228. source.Subscribe_(new AnonymousObserver<T>(onNext, onError, onCompleted), token);
  229. }
  230. private static void Subscribe_<T>(this IObservable<T> source, IObserver<T> observer, CancellationToken token)
  231. {
  232. if (token.CanBeCanceled)
  233. {
  234. if (!token.IsCancellationRequested)
  235. {
  236. var r = new SingleAssignmentDisposable();
  237. //
  238. // [OK] Use of unsafe Subscribe: exception during Subscribe doesn't orphan CancellationTokenRegistration.
  239. //
  240. var d = source.Subscribe/*Unsafe*/(
  241. observer.OnNext,
  242. ex =>
  243. {
  244. using (r)
  245. observer.OnError(ex);
  246. },
  247. () =>
  248. {
  249. using (r)
  250. observer.OnCompleted();
  251. }
  252. );
  253. r.Disposable = token.Register(d.Dispose);
  254. }
  255. }
  256. else
  257. {
  258. source.Subscribe(observer);
  259. }
  260. }
  261. #endif
  262. #endregion
  263. #region SubscribeSafe
  264. /// <summary>
  265. /// Subscribes to the specified source, re-routing synchronous exceptions during invocation of the Subscribe method to the observer's OnError channel.
  266. /// This method is typically used when writing query operators.
  267. /// </summary>
  268. /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
  269. /// <param name="source">Observable sequence to subscribe to.</param>
  270. /// <param name="observer">Observer that will be passed to the observable sequence, and that will be used for exception propagation.</param>
  271. /// <returns>IDisposable object used to unsubscribe from the observable sequence.</returns>
  272. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="observer"/> is null.</exception>
  273. [EditorBrowsable(EditorBrowsableState.Advanced)]
  274. public static IDisposable SubscribeSafe<T>(this IObservable<T> source, IObserver<T> observer)
  275. {
  276. if (source == null)
  277. throw new ArgumentNullException("source");
  278. if (observer == null)
  279. throw new ArgumentNullException("observer");
  280. //
  281. // The following types are white-listed and should not exhibit exceptional behavior
  282. // for regular operation circumstances.
  283. //
  284. if (source is ObservableBase<T>)
  285. return source.Subscribe(observer);
  286. #if !NO_PERF
  287. var producer = source as IProducer<T>;
  288. if (producer != null)
  289. return producer.SubscribeRaw(observer, false);
  290. #endif
  291. var d = Disposable.Empty;
  292. try
  293. {
  294. d = source.Subscribe(observer);
  295. }
  296. catch (Exception exception)
  297. {
  298. //
  299. // The effect of redirecting the exception to the OnError channel is automatic
  300. // clean-up of query operator state for a large number of cases. For example,
  301. // consider a binary and temporal query operator with the following Subscribe
  302. // behavior (implemented using the Producer pattern with a Run method):
  303. //
  304. // public IDisposable Run(...)
  305. // {
  306. // var tm = _scheduler.Schedule(_due, Tick);
  307. //
  308. // var df = _fst.SubscribeSafe(new FstObserver(this, ...));
  309. // var ds = _snd.SubscribeSafe(new SndObserver(this, ...)); // <-- fails
  310. //
  311. // return new CompositeDisposable(tm, df, ds);
  312. // }
  313. //
  314. // If the second subscription fails, we're not leaving the first subscription
  315. // or the scheduled job hanging around. Instead, the OnError propagation to
  316. // the SndObserver should take care of a Dispose call to the observer's parent
  317. // object. The handshake between Producer and Sink objects will ultimately
  318. // cause disposal of the CompositeDisposable that's returned from the method.
  319. //
  320. observer.OnError(exception);
  321. }
  322. return d;
  323. }
  324. #endregion
  325. }
  326. }