1
0

TaskObservableExtensions.cs 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367
  1. // this code is borrowed from RxOfficial(rx.codeplex.com) and modified
  2. #if (NET_4_6 || NET_STANDARD_2_0)
  3. using System;
  4. using System.Threading.Tasks;
  5. using System.Threading;
  6. namespace UniRx
  7. {
  8. /// <summary>
  9. /// Provides a set of static methods for converting tasks to observable sequences.
  10. /// </summary>
  11. public static class TaskObservableExtensions
  12. {
  13. /// <summary>
  14. /// Returns an observable sequence that signals when the task completes.
  15. /// </summary>
  16. /// <param name="task">Task to convert to an observable sequence.</param>
  17. /// <returns>An observable sequence that produces a unit value when the task completes, or propagates the exception produced by the task.</returns>
  18. /// <exception cref="ArgumentNullException"><paramref name="task"/> is null.</exception>
  19. /// <remarks>If the specified task object supports cancellation, consider using <see cref="Observable.FromAsync(Func{CancellationToken, Task})"/> instead.</remarks>
  20. public static IObservable<Unit> ToObservable(this Task task)
  21. {
  22. if (task == null)
  23. throw new ArgumentNullException("task");
  24. return ToObservableImpl(task, null);
  25. }
  26. /// <summary>
  27. /// Returns an observable sequence that signals when the task completes.
  28. /// </summary>
  29. /// <param name="task">Task to convert to an observable sequence.</param>
  30. /// <param name="scheduler">Scheduler on which to notify observers about completion, cancellation or failure.</param>
  31. /// <returns>An observable sequence that produces a unit value when the task completes, or propagates the exception produced by the task.</returns>
  32. /// <exception cref="ArgumentNullException"><paramref name="task"/> is null or <paramref name="scheduler"/> is null.</exception>
  33. /// <remarks>If the specified task object supports cancellation, consider using <see cref="Observable.FromAsync(Func{CancellationToken, Task})"/> instead.</remarks>
  34. public static IObservable<Unit> ToObservable(this Task task, IScheduler scheduler)
  35. {
  36. if (task == null)
  37. throw new ArgumentNullException("task");
  38. if (scheduler == null)
  39. throw new ArgumentNullException("scheduler");
  40. return ToObservableImpl(task, scheduler);
  41. }
  42. private static IObservable<Unit> ToObservableImpl(Task task, IScheduler scheduler)
  43. {
  44. var res = default(IObservable<Unit>);
  45. if (task.IsCompleted)
  46. {
  47. scheduler = scheduler ?? Scheduler.Immediate;
  48. switch (task.Status)
  49. {
  50. case TaskStatus.RanToCompletion:
  51. res = Observable.Return<Unit>(Unit.Default, scheduler);
  52. break;
  53. case TaskStatus.Faulted:
  54. res = Observable.Throw<Unit>(task.Exception.InnerException, scheduler);
  55. break;
  56. case TaskStatus.Canceled:
  57. res = Observable.Throw<Unit>(new TaskCanceledException(task), scheduler);
  58. break;
  59. }
  60. }
  61. else
  62. {
  63. //
  64. // Separate method to avoid closure in synchronous completion case.
  65. //
  66. res = ToObservableSlow(task, scheduler);
  67. }
  68. return res;
  69. }
  70. private static IObservable<Unit> ToObservableSlow(Task task, IScheduler scheduler)
  71. {
  72. var subject = new AsyncSubject<Unit>();
  73. var options = GetTaskContinuationOptions(scheduler);
  74. task.ContinueWith(t => ToObservableDone(task, subject), options);
  75. return ToObservableResult(subject, scheduler);
  76. }
  77. private static void ToObservableDone(Task task, IObserver<Unit> subject)
  78. {
  79. switch (task.Status)
  80. {
  81. case TaskStatus.RanToCompletion:
  82. subject.OnNext(Unit.Default);
  83. subject.OnCompleted();
  84. break;
  85. case TaskStatus.Faulted:
  86. subject.OnError(task.Exception.InnerException);
  87. break;
  88. case TaskStatus.Canceled:
  89. subject.OnError(new TaskCanceledException(task));
  90. break;
  91. }
  92. }
  93. /// <summary>
  94. /// Returns an observable sequence that propagates the result of the task.
  95. /// </summary>
  96. /// <typeparam name="TResult">The type of the result produced by the task.</typeparam>
  97. /// <param name="task">Task to convert to an observable sequence.</param>
  98. /// <returns>An observable sequence that produces the task's result, or propagates the exception produced by the task.</returns>
  99. /// <exception cref="ArgumentNullException"><paramref name="task"/> is null.</exception>
  100. /// <remarks>If the specified task object supports cancellation, consider using <see cref="Observable.FromAsync{TResult}(Func{CancellationToken, Task{TResult}})"/> instead.</remarks>
  101. public static IObservable<TResult> ToObservable<TResult>(this Task<TResult> task)
  102. {
  103. if (task == null)
  104. throw new ArgumentNullException("task");
  105. return ToObservableImpl(task, null);
  106. }
  107. /// <summary>
  108. /// Returns an observable sequence that propagates the result of the task.
  109. /// </summary>
  110. /// <typeparam name="TResult">The type of the result produced by the task.</typeparam>
  111. /// <param name="task">Task to convert to an observable sequence.</param>
  112. /// <param name="scheduler">Scheduler on which to notify observers about completion, cancellation or failure.</param>
  113. /// <returns>An observable sequence that produces the task's result, or propagates the exception produced by the task.</returns>
  114. /// <exception cref="ArgumentNullException"><paramref name="task"/> is null or <paramref name="scheduler"/> is null.</exception>
  115. /// <remarks>If the specified task object supports cancellation, consider using <see cref="Observable.FromAsync{TResult}(Func{CancellationToken, Task{TResult}})"/> instead.</remarks>
  116. public static IObservable<TResult> ToObservable<TResult>(this Task<TResult> task, IScheduler scheduler)
  117. {
  118. if (task == null)
  119. throw new ArgumentNullException("task");
  120. if (scheduler == null)
  121. throw new ArgumentNullException("scheduler");
  122. return ToObservableImpl(task, scheduler);
  123. }
  124. private static IObservable<TResult> ToObservableImpl<TResult>(Task<TResult> task, IScheduler scheduler)
  125. {
  126. var res = default(IObservable<TResult>);
  127. if (task.IsCompleted)
  128. {
  129. scheduler = scheduler ?? Scheduler.Immediate;
  130. switch (task.Status)
  131. {
  132. case TaskStatus.RanToCompletion:
  133. res = Observable.Return<TResult>(task.Result, scheduler);
  134. break;
  135. case TaskStatus.Faulted:
  136. res = Observable.Throw<TResult>(task.Exception.InnerException, scheduler);
  137. break;
  138. case TaskStatus.Canceled:
  139. res = Observable.Throw<TResult>(new TaskCanceledException(task), scheduler);
  140. break;
  141. }
  142. }
  143. else
  144. {
  145. //
  146. // Separate method to avoid closure in synchronous completion case.
  147. //
  148. res = ToObservableSlow(task, scheduler);
  149. }
  150. return res;
  151. }
  152. private static IObservable<TResult> ToObservableSlow<TResult>(Task<TResult> task, IScheduler scheduler)
  153. {
  154. var subject = new AsyncSubject<TResult>();
  155. var options = GetTaskContinuationOptions(scheduler);
  156. task.ContinueWith(t => ToObservableDone(task, subject), options);
  157. return ToObservableResult(subject, scheduler);
  158. }
  159. private static void ToObservableDone<TResult>(Task<TResult> task, IObserver<TResult> subject)
  160. {
  161. switch (task.Status)
  162. {
  163. case TaskStatus.RanToCompletion:
  164. subject.OnNext(task.Result);
  165. subject.OnCompleted();
  166. break;
  167. case TaskStatus.Faulted:
  168. subject.OnError(task.Exception.InnerException);
  169. break;
  170. case TaskStatus.Canceled:
  171. subject.OnError(new TaskCanceledException(task));
  172. break;
  173. }
  174. }
  175. private static TaskContinuationOptions GetTaskContinuationOptions(IScheduler scheduler)
  176. {
  177. var options = TaskContinuationOptions.None;
  178. if (scheduler != null)
  179. {
  180. //
  181. // We explicitly don't special-case the immediate scheduler here. If the user asks for a
  182. // synchronous completion, we'll try our best. However, there's no guarantee due to the
  183. // internal stack probing in the TPL, which may cause asynchronous completion on a thread
  184. // pool thread in order to avoid stack overflows. Therefore we can only attempt to be more
  185. // efficient in the case where the user specified a scheduler, hence we know that the
  186. // continuation will trigger a scheduling operation. In case of the immediate scheduler,
  187. // it really becomes "immediate scheduling" wherever the TPL decided to run the continuation,
  188. // i.e. not necessarily where the task was completed from.
  189. //
  190. options |= TaskContinuationOptions.ExecuteSynchronously;
  191. }
  192. return options;
  193. }
  194. private static IObservable<TResult> ToObservableResult<TResult>(AsyncSubject<TResult> subject, IScheduler scheduler)
  195. {
  196. if (scheduler != null)
  197. {
  198. return subject.ObserveOn(scheduler);
  199. }
  200. else
  201. {
  202. return subject.AsObservable();
  203. }
  204. }
  205. /// <summary>
  206. /// Returns a task that will receive the last value or the exception produced by the observable sequence.
  207. /// </summary>
  208. /// <typeparam name="TResult">The type of the elements in the source sequence.</typeparam>
  209. /// <param name="observable">Observable sequence to convert to a task.</param>
  210. /// <returns>A task that will receive the last element or the exception produced by the observable sequence.</returns>
  211. /// <exception cref="ArgumentNullException"><paramref name="observable"/> is null.</exception>
  212. public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable)
  213. {
  214. if (observable == null)
  215. throw new ArgumentNullException("observable");
  216. return observable.ToTask(new CancellationToken(), null);
  217. }
  218. /// <summary>
  219. /// Returns a task that will receive the last value or the exception produced by the observable sequence.
  220. /// </summary>
  221. /// <typeparam name="TResult">The type of the elements in the source sequence.</typeparam>
  222. /// <param name="observable">Observable sequence to convert to a task.</param>
  223. /// <param name="state">The state to use as the underlying task's AsyncState.</param>
  224. /// <returns>A task that will receive the last element or the exception produced by the observable sequence.</returns>
  225. /// <exception cref="ArgumentNullException"><paramref name="observable"/> is null.</exception>
  226. public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable, object state)
  227. {
  228. if (observable == null)
  229. throw new ArgumentNullException("observable");
  230. return observable.ToTask(new CancellationToken(), state);
  231. }
  232. /// <summary>
  233. /// Returns a task that will receive the last value or the exception produced by the observable sequence.
  234. /// </summary>
  235. /// <typeparam name="TResult">The type of the elements in the source sequence.</typeparam>
  236. /// <param name="observable">Observable sequence to convert to a task.</param>
  237. /// <param name="cancellationToken">Cancellation token that can be used to cancel the task, causing unsubscription from the observable sequence.</param>
  238. /// <returns>A task that will receive the last element or the exception produced by the observable sequence.</returns>
  239. /// <exception cref="ArgumentNullException"><paramref name="observable"/> is null.</exception>
  240. public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable, CancellationToken cancellationToken)
  241. {
  242. if (observable == null)
  243. throw new ArgumentNullException("observable");
  244. return observable.ToTask(cancellationToken, null);
  245. }
  246. /// <summary>
  247. /// Returns a task that will receive the last value or the exception produced by the observable sequence.
  248. /// </summary>
  249. /// <typeparam name="TResult">The type of the elements in the source sequence.</typeparam>
  250. /// <param name="observable">Observable sequence to convert to a task.</param>
  251. /// <param name="cancellationToken">Cancellation token that can be used to cancel the task, causing unsubscription from the observable sequence.</param>
  252. /// <param name="state">The state to use as the underlying task's AsyncState.</param>
  253. /// <returns>A task that will receive the last element or the exception produced by the observable sequence.</returns>
  254. /// <exception cref="ArgumentNullException"><paramref name="observable"/> is null.</exception>
  255. public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable, CancellationToken cancellationToken, object state)
  256. {
  257. if (observable == null)
  258. throw new ArgumentNullException("observable");
  259. var hasValue = false;
  260. var lastValue = default(TResult);
  261. var tcs = new TaskCompletionSource<TResult>(state);
  262. var disposable = new SingleAssignmentDisposable();
  263. var ctr = default(CancellationTokenRegistration);
  264. if (cancellationToken.CanBeCanceled)
  265. {
  266. ctr = cancellationToken.Register(() =>
  267. {
  268. disposable.Dispose();
  269. tcs.TrySetCanceled(cancellationToken);
  270. });
  271. }
  272. var taskCompletionObserver = Observer.Create<TResult>(
  273. value =>
  274. {
  275. hasValue = true;
  276. lastValue = value;
  277. },
  278. ex =>
  279. {
  280. tcs.TrySetException(ex);
  281. ctr.Dispose(); // no null-check needed (struct)
  282. disposable.Dispose();
  283. },
  284. () =>
  285. {
  286. if (hasValue)
  287. tcs.TrySetResult(lastValue);
  288. else
  289. tcs.TrySetException(new InvalidOperationException("Strings_Linq.NO_ELEMENTS"));
  290. ctr.Dispose(); // no null-check needed (struct)
  291. disposable.Dispose();
  292. }
  293. );
  294. //
  295. // Subtle race condition: if the source completes before we reach the line below, the SingleAssigmentDisposable
  296. // will already have been disposed. Upon assignment, the disposable resource being set will be disposed on the
  297. // spot, which may throw an exception. (Similar to TFS 487142)
  298. //
  299. try
  300. {
  301. //
  302. // [OK] Use of unsafe Subscribe: we're catching the exception here to set the TaskCompletionSource.
  303. //
  304. // Notice we could use a safe subscription to route errors through OnError, but we still need the
  305. // exception handling logic here for the reason explained above. We cannot afford to throw here
  306. // and as a result never set the TaskCompletionSource, so we tunnel everything through here.
  307. //
  308. disposable.Disposable = observable.Subscribe/*Unsafe*/(taskCompletionObserver);
  309. }
  310. catch (Exception ex)
  311. {
  312. tcs.TrySetException(ex);
  313. }
  314. return tcs.Task;
  315. }
  316. }
  317. }
  318. #endif