123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133 |
- #if (NET_4_6 || NET_STANDARD_2_0)
- using System;
- using System.Collections.Generic;
- using System.Linq;
- using System.Text;
- using System.Threading;
- using System.Threading.Tasks;
- namespace UniRx
- {
- public static partial class Observable
- {
-
-
-
-
-
- public static AsyncSubject<TSource> GetAwaiter<TSource>(this IObservable<TSource> source)
- {
- if (source == null) throw new ArgumentNullException("source");
-
- return RunAsync(source, CancellationToken.None);
- }
-
-
-
-
-
- public static AsyncSubject<TSource> GetAwaiter<TSource>(this IConnectableObservable<TSource> source)
- {
- if (source == null) throw new ArgumentNullException("source");
- return RunAsync(source, CancellationToken.None);
- }
-
-
-
-
-
-
- public static AsyncSubject<TSource> GetAwaiter<TSource>(this IObservable<TSource> source, CancellationToken cancellationToken)
- {
- if (source == null) throw new ArgumentNullException("source");
- return RunAsync(source, cancellationToken);
- }
-
-
-
-
-
-
- public static AsyncSubject<TSource> GetAwaiter<TSource>(this IConnectableObservable<TSource> source, CancellationToken cancellationToken)
- {
- if (source == null) throw new ArgumentNullException("source");
- return RunAsync(source, cancellationToken);
- }
- static AsyncSubject<TSource> RunAsync<TSource>(IObservable<TSource> source, CancellationToken cancellationToken)
- {
- var s = new AsyncSubject<TSource>();
- if (cancellationToken.IsCancellationRequested)
- {
- return Cancel(s, cancellationToken);
- }
- var d = source.Subscribe(s);
- if (cancellationToken.CanBeCanceled)
- {
- RegisterCancelation(s, d, cancellationToken);
- }
- return s;
- }
- static AsyncSubject<TSource> RunAsync<TSource>(IConnectableObservable<TSource> source, CancellationToken cancellationToken)
- {
- var s = new AsyncSubject<TSource>();
- if (cancellationToken.IsCancellationRequested)
- {
- return Cancel(s, cancellationToken);
- }
- var d = source.Subscribe(s);
- var c = source.Connect();
- if (cancellationToken.CanBeCanceled)
- {
- RegisterCancelation(s, StableCompositeDisposable.Create(d, c), cancellationToken);
- }
- return s;
- }
- static AsyncSubject<T> Cancel<T>(AsyncSubject<T> subject, CancellationToken cancellationToken)
- {
- subject.OnError(new OperationCanceledException(cancellationToken));
- return subject;
- }
- static void RegisterCancelation<T>(AsyncSubject<T> subject, IDisposable subscription, CancellationToken token)
- {
-
-
-
-
- var ctr = token.Register(() =>
- {
- subscription.Dispose();
- Cancel(subject, token);
- });
-
-
-
-
-
-
- subject.Subscribe(Stubs<T>.Ignore, _ => ctr.Dispose(), ctr.Dispose);
- }
- }
- }
- #endif
|