Observable.cs 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292
  1. using System;
  2. using System.Collections;
  3. using System.Collections.Generic;
  4. using System.Threading;
  5. using UniRx.InternalUtil;
  6. using UniRx.Operators;
  7. namespace UniRx
  8. {
  9. // Standard Query Operators
  10. // onNext implementation guide. enclose otherFunc but onNext is not catch.
  11. // try{ otherFunc(); } catch { onError() }
  12. // onNext();
  13. public static partial class Observable
  14. {
  15. static readonly TimeSpan InfiniteTimeSpan = new TimeSpan(0, 0, 0, 0, -1); // from .NET 4.5
  16. public static IObservable<TR> Select<T, TR>(this IObservable<T> source, Func<T, TR> selector)
  17. {
  18. // sometimes cause "which no ahead of time (AOT) code was generated." on IL2CPP...
  19. //var select = source as ISelect<T>;
  20. //if (select != null)
  21. //{
  22. // return select.CombineSelector(selector);
  23. //}
  24. // optimized path
  25. var whereObservable = source as UniRx.Operators.WhereObservable<T>;
  26. if (whereObservable != null)
  27. {
  28. return whereObservable.CombineSelector<TR>(selector);
  29. }
  30. return new SelectObservable<T, TR>(source, selector);
  31. }
  32. public static IObservable<TR> Select<T, TR>(this IObservable<T> source, Func<T, int, TR> selector)
  33. {
  34. return new SelectObservable<T, TR>(source, selector);
  35. }
  36. public static IObservable<T> Where<T>(this IObservable<T> source, Func<T, bool> predicate)
  37. {
  38. // optimized path
  39. var whereObservable = source as UniRx.Operators.WhereObservable<T>;
  40. if (whereObservable != null)
  41. {
  42. return whereObservable.CombinePredicate(predicate);
  43. }
  44. var selectObservable = source as UniRx.Operators.ISelect<T>;
  45. if (selectObservable != null)
  46. {
  47. return selectObservable.CombinePredicate(predicate);
  48. }
  49. return new WhereObservable<T>(source, predicate);
  50. }
  51. public static IObservable<T> Where<T>(this IObservable<T> source, Func<T, int, bool> predicate)
  52. {
  53. return new WhereObservable<T>(source, predicate);
  54. }
  55. /// <summary>
  56. /// Lightweight SelectMany for Single Async Operation.
  57. /// </summary>
  58. public static IObservable<TR> ContinueWith<T, TR>(this IObservable<T> source, IObservable<TR> other)
  59. {
  60. return ContinueWith(source, _ => other);
  61. }
  62. /// <summary>
  63. /// Lightweight SelectMany for Single Async Operation.
  64. /// </summary>
  65. public static IObservable<TR> ContinueWith<T, TR>(this IObservable<T> source, Func<T, IObservable<TR>> selector)
  66. {
  67. return new ContinueWithObservable<T, TR>(source, selector);
  68. }
  69. public static IObservable<TR> SelectMany<T, TR>(this IObservable<T> source, IObservable<TR> other)
  70. {
  71. return SelectMany(source, _ => other);
  72. }
  73. public static IObservable<TR> SelectMany<T, TR>(this IObservable<T> source, Func<T, IObservable<TR>> selector)
  74. {
  75. return new SelectManyObservable<T, TR>(source, selector);
  76. }
  77. public static IObservable<TResult> SelectMany<TSource, TResult>(this IObservable<TSource> source, Func<TSource, int, IObservable<TResult>> selector)
  78. {
  79. return new SelectManyObservable<TSource, TResult>(source, selector);
  80. }
  81. public static IObservable<TR> SelectMany<T, TC, TR>(this IObservable<T> source, Func<T, IObservable<TC>> collectionSelector, Func<T, TC, TR> resultSelector)
  82. {
  83. return new SelectManyObservable<T, TC, TR>(source, collectionSelector, resultSelector);
  84. }
  85. public static IObservable<TResult> SelectMany<TSource, TCollection, TResult>(this IObservable<TSource> source, Func<TSource, int, IObservable<TCollection>> collectionSelector, Func<TSource, int, TCollection, int, TResult> resultSelector)
  86. {
  87. return new SelectManyObservable<TSource, TCollection, TResult>(source, collectionSelector, resultSelector);
  88. }
  89. public static IObservable<TResult> SelectMany<TSource, TResult>(this IObservable<TSource> source, Func<TSource, IEnumerable<TResult>> selector)
  90. {
  91. return new SelectManyObservable<TSource, TResult>(source, selector);
  92. }
  93. public static IObservable<TResult> SelectMany<TSource, TResult>(this IObservable<TSource> source, Func<TSource, int, IEnumerable<TResult>> selector)
  94. {
  95. return new SelectManyObservable<TSource, TResult>(source, selector);
  96. }
  97. public static IObservable<TResult> SelectMany<TSource, TCollection, TResult>(this IObservable<TSource> source, Func<TSource, IEnumerable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector)
  98. {
  99. return new SelectManyObservable<TSource, TCollection, TResult>(source, collectionSelector, resultSelector);
  100. }
  101. public static IObservable<TResult> SelectMany<TSource, TCollection, TResult>(this IObservable<TSource> source, Func<TSource, int, IEnumerable<TCollection>> collectionSelector, Func<TSource, int, TCollection, int, TResult> resultSelector)
  102. {
  103. return new SelectManyObservable<TSource, TCollection, TResult>(source, collectionSelector, resultSelector);
  104. }
  105. public static IObservable<T[]> ToArray<T>(this IObservable<T> source)
  106. {
  107. return new ToArrayObservable<T>(source);
  108. }
  109. public static IObservable<IList<T>> ToList<T>(this IObservable<T> source)
  110. {
  111. return new ToListObservable<T>(source);
  112. }
  113. public static IObservable<T> Do<T>(this IObservable<T> source, IObserver<T> observer)
  114. {
  115. return new DoObserverObservable<T>(source, observer);
  116. }
  117. public static IObservable<T> Do<T>(this IObservable<T> source, Action<T> onNext)
  118. {
  119. return new DoObservable<T>(source, onNext, Stubs.Throw, Stubs.Nop);
  120. }
  121. public static IObservable<T> Do<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError)
  122. {
  123. return new DoObservable<T>(source, onNext, onError, Stubs.Nop);
  124. }
  125. public static IObservable<T> Do<T>(this IObservable<T> source, Action<T> onNext, Action onCompleted)
  126. {
  127. return new DoObservable<T>(source, onNext, Stubs.Throw, onCompleted);
  128. }
  129. public static IObservable<T> Do<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError, Action onCompleted)
  130. {
  131. return new DoObservable<T>(source, onNext, onError, onCompleted);
  132. }
  133. public static IObservable<T> DoOnError<T>(this IObservable<T> source, Action<Exception> onError)
  134. {
  135. return new DoOnErrorObservable<T>(source, onError);
  136. }
  137. public static IObservable<T> DoOnCompleted<T>(this IObservable<T> source, Action onCompleted)
  138. {
  139. return new DoOnCompletedObservable<T>(source, onCompleted);
  140. }
  141. public static IObservable<T> DoOnTerminate<T>(this IObservable<T> source, Action onTerminate)
  142. {
  143. return new DoOnTerminateObservable<T>(source, onTerminate);
  144. }
  145. public static IObservable<T> DoOnSubscribe<T>(this IObservable<T> source, Action onSubscribe)
  146. {
  147. return new DoOnSubscribeObservable<T>(source, onSubscribe);
  148. }
  149. public static IObservable<T> DoOnCancel<T>(this IObservable<T> source, Action onCancel)
  150. {
  151. return new DoOnCancelObservable<T>(source, onCancel);
  152. }
  153. public static IObservable<Notification<T>> Materialize<T>(this IObservable<T> source)
  154. {
  155. return new MaterializeObservable<T>(source);
  156. }
  157. public static IObservable<T> Dematerialize<T>(this IObservable<Notification<T>> source)
  158. {
  159. return new DematerializeObservable<T>(source);
  160. }
  161. public static IObservable<T> DefaultIfEmpty<T>(this IObservable<T> source)
  162. {
  163. return new DefaultIfEmptyObservable<T>(source, default(T));
  164. }
  165. public static IObservable<T> DefaultIfEmpty<T>(this IObservable<T> source, T defaultValue)
  166. {
  167. return new DefaultIfEmptyObservable<T>(source, defaultValue);
  168. }
  169. public static IObservable<TSource> Distinct<TSource>(this IObservable<TSource> source)
  170. {
  171. #if !UniRxLibrary
  172. var comparer = UnityEqualityComparer.GetDefault<TSource>();
  173. #else
  174. var comparer = EqualityComparer<TSource>.Default;
  175. #endif
  176. return new DistinctObservable<TSource>(source, comparer);
  177. }
  178. public static IObservable<TSource> Distinct<TSource>(this IObservable<TSource> source, IEqualityComparer<TSource> comparer)
  179. {
  180. return new DistinctObservable<TSource>(source, comparer);
  181. }
  182. public static IObservable<TSource> Distinct<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector)
  183. {
  184. #if !UniRxLibrary
  185. var comparer = UnityEqualityComparer.GetDefault<TKey>();
  186. #else
  187. var comparer = EqualityComparer<TKey>.Default;
  188. #endif
  189. return new DistinctObservable<TSource, TKey>(source, keySelector, comparer);
  190. }
  191. public static IObservable<TSource> Distinct<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> comparer)
  192. {
  193. return new DistinctObservable<TSource, TKey>(source, keySelector, comparer);
  194. }
  195. public static IObservable<T> DistinctUntilChanged<T>(this IObservable<T> source)
  196. {
  197. #if !UniRxLibrary
  198. var comparer = UnityEqualityComparer.GetDefault<T>();
  199. #else
  200. var comparer = EqualityComparer<T>.Default;
  201. #endif
  202. return new DistinctUntilChangedObservable<T>(source, comparer);
  203. }
  204. public static IObservable<T> DistinctUntilChanged<T>(this IObservable<T> source, IEqualityComparer<T> comparer)
  205. {
  206. if (source == null) throw new ArgumentNullException("source");
  207. return new DistinctUntilChangedObservable<T>(source, comparer);
  208. }
  209. public static IObservable<T> DistinctUntilChanged<T, TKey>(this IObservable<T> source, Func<T, TKey> keySelector)
  210. {
  211. #if !UniRxLibrary
  212. var comparer = UnityEqualityComparer.GetDefault<TKey>();
  213. #else
  214. var comparer = EqualityComparer<TKey>.Default;
  215. #endif
  216. return new DistinctUntilChangedObservable<T, TKey>(source, keySelector, comparer);
  217. }
  218. public static IObservable<T> DistinctUntilChanged<T, TKey>(this IObservable<T> source, Func<T, TKey> keySelector, IEqualityComparer<TKey> comparer)
  219. {
  220. if (source == null) throw new ArgumentNullException("source");
  221. return new DistinctUntilChangedObservable<T, TKey>(source, keySelector, comparer);
  222. }
  223. public static IObservable<T> IgnoreElements<T>(this IObservable<T> source)
  224. {
  225. return new IgnoreElementsObservable<T>(source);
  226. }
  227. public static IObservable<Unit> ForEachAsync<T>(this IObservable<T> source, Action<T> onNext)
  228. {
  229. return new ForEachAsyncObservable<T>(source, onNext);
  230. }
  231. public static IObservable<Unit> ForEachAsync<T>(this IObservable<T> source, Action<T, int> onNext)
  232. {
  233. return new ForEachAsyncObservable<T>(source, onNext);
  234. }
  235. }
  236. }