Observable.Creation.cs 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375
  1. using System;
  2. using System.Collections.Generic;
  3. using UniRx.Operators;
  4. namespace UniRx
  5. {
  6. public static partial class Observable
  7. {
  8. /// <summary>
  9. /// Create anonymous observable. Observer has exception durability. This is recommended for make operator and event like generator.
  10. /// </summary>
  11. public static IObservable<T> Create<T>(Func<IObserver<T>, IDisposable> subscribe)
  12. {
  13. if (subscribe == null) throw new ArgumentNullException("subscribe");
  14. return new CreateObservable<T>(subscribe);
  15. }
  16. /// <summary>
  17. /// Create anonymous observable. Observer has exception durability. This is recommended for make operator and event like generator(HotObservable).
  18. /// </summary>
  19. public static IObservable<T> Create<T>(Func<IObserver<T>, IDisposable> subscribe, bool isRequiredSubscribeOnCurrentThread)
  20. {
  21. if (subscribe == null) throw new ArgumentNullException("subscribe");
  22. return new CreateObservable<T>(subscribe, isRequiredSubscribeOnCurrentThread);
  23. }
  24. /// <summary>
  25. /// Create anonymous observable. Observer has exception durability. This is recommended for make operator and event like generator.
  26. /// </summary>
  27. public static IObservable<T> CreateWithState<T, TState>(TState state, Func<TState, IObserver<T>, IDisposable> subscribe)
  28. {
  29. if (subscribe == null) throw new ArgumentNullException("subscribe");
  30. return new CreateObservable<T, TState>(state, subscribe);
  31. }
  32. /// <summary>
  33. /// Create anonymous observable. Observer has exception durability. This is recommended for make operator and event like generator(HotObservable).
  34. /// </summary>
  35. public static IObservable<T> CreateWithState<T, TState>(TState state, Func<TState, IObserver<T>, IDisposable> subscribe, bool isRequiredSubscribeOnCurrentThread)
  36. {
  37. if (subscribe == null) throw new ArgumentNullException("subscribe");
  38. return new CreateObservable<T, TState>(state, subscribe, isRequiredSubscribeOnCurrentThread);
  39. }
  40. /// <summary>
  41. /// Create anonymous observable. Safe means auto detach when error raised in onNext pipeline. This is recommended for make generator (ColdObservable).
  42. /// </summary>
  43. public static IObservable<T> CreateSafe<T>(Func<IObserver<T>, IDisposable> subscribe)
  44. {
  45. if (subscribe == null) throw new ArgumentNullException("subscribe");
  46. return new CreateSafeObservable<T>(subscribe);
  47. }
  48. /// <summary>
  49. /// Create anonymous observable. Safe means auto detach when error raised in onNext pipeline. This is recommended for make generator (ColdObservable).
  50. /// </summary>
  51. public static IObservable<T> CreateSafe<T>(Func<IObserver<T>, IDisposable> subscribe, bool isRequiredSubscribeOnCurrentThread)
  52. {
  53. if (subscribe == null) throw new ArgumentNullException("subscribe");
  54. return new CreateSafeObservable<T>(subscribe, isRequiredSubscribeOnCurrentThread);
  55. }
  56. /// <summary>
  57. /// Empty Observable. Returns only OnCompleted.
  58. /// </summary>
  59. public static IObservable<T> Empty<T>()
  60. {
  61. return Empty<T>(Scheduler.DefaultSchedulers.ConstantTimeOperations);
  62. }
  63. /// <summary>
  64. /// Empty Observable. Returns only OnCompleted on specified scheduler.
  65. /// </summary>
  66. public static IObservable<T> Empty<T>(IScheduler scheduler)
  67. {
  68. if (scheduler == Scheduler.Immediate)
  69. {
  70. return ImmutableEmptyObservable<T>.Instance;
  71. }
  72. else
  73. {
  74. return new EmptyObservable<T>(scheduler);
  75. }
  76. }
  77. /// <summary>
  78. /// Empty Observable. Returns only OnCompleted. witness is for type inference.
  79. /// </summary>
  80. public static IObservable<T> Empty<T>(T witness)
  81. {
  82. return Empty<T>(Scheduler.DefaultSchedulers.ConstantTimeOperations);
  83. }
  84. /// <summary>
  85. /// Empty Observable. Returns only OnCompleted on specified scheduler. witness is for type inference.
  86. /// </summary>
  87. public static IObservable<T> Empty<T>(IScheduler scheduler, T witness)
  88. {
  89. return Empty<T>(scheduler);
  90. }
  91. /// <summary>
  92. /// Non-Terminating Observable. It's no returns, never finish.
  93. /// </summary>
  94. public static IObservable<T> Never<T>()
  95. {
  96. return ImmutableNeverObservable<T>.Instance;
  97. }
  98. /// <summary>
  99. /// Non-Terminating Observable. It's no returns, never finish. witness is for type inference.
  100. /// </summary>
  101. public static IObservable<T> Never<T>(T witness)
  102. {
  103. return ImmutableNeverObservable<T>.Instance;
  104. }
  105. /// <summary>
  106. /// Return single sequence Immediately.
  107. /// </summary>
  108. public static IObservable<T> Return<T>(T value)
  109. {
  110. return Return<T>(value, Scheduler.DefaultSchedulers.ConstantTimeOperations);
  111. }
  112. /// <summary>
  113. /// Return single sequence on specified scheduler.
  114. /// </summary>
  115. public static IObservable<T> Return<T>(T value, IScheduler scheduler)
  116. {
  117. if (scheduler == Scheduler.Immediate)
  118. {
  119. return new ImmediateReturnObservable<T>(value);
  120. }
  121. else
  122. {
  123. return new ReturnObservable<T>(value, scheduler);
  124. }
  125. }
  126. /// <summary>
  127. /// Return single sequence Immediately, optimized for Unit(no allocate memory).
  128. /// </summary>
  129. public static IObservable<Unit> Return(Unit value)
  130. {
  131. return ImmutableReturnUnitObservable.Instance;
  132. }
  133. /// <summary>
  134. /// Return single sequence Immediately, optimized for Boolean(no allocate memory).
  135. /// </summary>
  136. public static IObservable<bool> Return(bool value)
  137. {
  138. return (value == true)
  139. ? (IObservable<bool>)ImmutableReturnTrueObservable.Instance
  140. : (IObservable<bool>)ImmutableReturnFalseObservable.Instance;
  141. }
  142. /// <summary>
  143. /// Return single sequence Immediately, optimized for Int32.
  144. /// </summary>
  145. public static IObservable<Int32> Return(int value)
  146. {
  147. return ImmutableReturnInt32Observable.GetInt32Observable(value);
  148. }
  149. /// <summary>
  150. /// Same as Observable.Return(Unit.Default); but no allocate memory.
  151. /// </summary>
  152. public static IObservable<Unit> ReturnUnit()
  153. {
  154. return ImmutableReturnUnitObservable.Instance;
  155. }
  156. /// <summary>
  157. /// Empty Observable. Returns only onError.
  158. /// </summary>
  159. public static IObservable<T> Throw<T>(Exception error)
  160. {
  161. return Throw<T>(error, Scheduler.DefaultSchedulers.ConstantTimeOperations);
  162. }
  163. /// <summary>
  164. /// Empty Observable. Returns only onError. witness if for Type inference.
  165. /// </summary>
  166. public static IObservable<T> Throw<T>(Exception error, T witness)
  167. {
  168. return Throw<T>(error, Scheduler.DefaultSchedulers.ConstantTimeOperations);
  169. }
  170. /// <summary>
  171. /// Empty Observable. Returns only onError on specified scheduler.
  172. /// </summary>
  173. public static IObservable<T> Throw<T>(Exception error, IScheduler scheduler)
  174. {
  175. return new ThrowObservable<T>(error, scheduler);
  176. }
  177. /// <summary>
  178. /// Empty Observable. Returns only onError on specified scheduler. witness if for Type inference.
  179. /// </summary>
  180. public static IObservable<T> Throw<T>(Exception error, IScheduler scheduler, T witness)
  181. {
  182. return Throw<T>(error, scheduler);
  183. }
  184. public static IObservable<int> Range(int start, int count)
  185. {
  186. return Range(start, count, Scheduler.DefaultSchedulers.Iteration);
  187. }
  188. public static IObservable<int> Range(int start, int count, IScheduler scheduler)
  189. {
  190. return new RangeObservable(start, count, scheduler);
  191. }
  192. public static IObservable<T> Repeat<T>(T value)
  193. {
  194. return Repeat(value, Scheduler.DefaultSchedulers.Iteration);
  195. }
  196. public static IObservable<T> Repeat<T>(T value, IScheduler scheduler)
  197. {
  198. if (scheduler == null) throw new ArgumentNullException("scheduler");
  199. return new RepeatObservable<T>(value, null, scheduler);
  200. }
  201. public static IObservable<T> Repeat<T>(T value, int repeatCount)
  202. {
  203. return Repeat(value, repeatCount, Scheduler.DefaultSchedulers.Iteration);
  204. }
  205. public static IObservable<T> Repeat<T>(T value, int repeatCount, IScheduler scheduler)
  206. {
  207. if (repeatCount < 0) throw new ArgumentOutOfRangeException("repeatCount");
  208. if (scheduler == null) throw new ArgumentNullException("scheduler");
  209. return new RepeatObservable<T>(value, repeatCount, scheduler);
  210. }
  211. public static IObservable<T> Repeat<T>(this IObservable<T> source)
  212. {
  213. return RepeatInfinite(source).Concat();
  214. }
  215. static IEnumerable<IObservable<T>> RepeatInfinite<T>(IObservable<T> source)
  216. {
  217. while (true)
  218. {
  219. yield return source;
  220. }
  221. }
  222. /// <summary>
  223. /// Same as Repeat() but if arriving contiguous "OnComplete" Repeat stops.
  224. /// </summary>
  225. public static IObservable<T> RepeatSafe<T>(this IObservable<T> source)
  226. {
  227. return new RepeatSafeObservable<T>(RepeatInfinite(source), source.IsRequiredSubscribeOnCurrentThread());
  228. }
  229. public static IObservable<T> Defer<T>(Func<IObservable<T>> observableFactory)
  230. {
  231. return new DeferObservable<T>(observableFactory);
  232. }
  233. public static IObservable<T> Start<T>(Func<T> function)
  234. {
  235. return new StartObservable<T>(function, null, Scheduler.DefaultSchedulers.AsyncConversions);
  236. }
  237. public static IObservable<T> Start<T>(Func<T> function, TimeSpan timeSpan)
  238. {
  239. return new StartObservable<T>(function, timeSpan, Scheduler.DefaultSchedulers.AsyncConversions);
  240. }
  241. public static IObservable<T> Start<T>(Func<T> function, IScheduler scheduler)
  242. {
  243. return new StartObservable<T>(function, null, scheduler);
  244. }
  245. public static IObservable<T> Start<T>(Func<T> function, TimeSpan timeSpan, IScheduler scheduler)
  246. {
  247. return new StartObservable<T>(function, timeSpan, scheduler);
  248. }
  249. public static IObservable<Unit> Start(Action action)
  250. {
  251. return new StartObservable<Unit>(action, null, Scheduler.DefaultSchedulers.AsyncConversions);
  252. }
  253. public static IObservable<Unit> Start(Action action, TimeSpan timeSpan)
  254. {
  255. return new StartObservable<Unit>(action, timeSpan, Scheduler.DefaultSchedulers.AsyncConversions);
  256. }
  257. public static IObservable<Unit> Start(Action action, IScheduler scheduler)
  258. {
  259. return new StartObservable<Unit>(action, null, scheduler);
  260. }
  261. public static IObservable<Unit> Start(Action action, TimeSpan timeSpan, IScheduler scheduler)
  262. {
  263. return new StartObservable<Unit>(action, timeSpan, scheduler);
  264. }
  265. public static Func<IObservable<T>> ToAsync<T>(Func<T> function)
  266. {
  267. return ToAsync(function, Scheduler.DefaultSchedulers.AsyncConversions);
  268. }
  269. public static Func<IObservable<T>> ToAsync<T>(Func<T> function, IScheduler scheduler)
  270. {
  271. return () =>
  272. {
  273. var subject = new AsyncSubject<T>();
  274. scheduler.Schedule(() =>
  275. {
  276. var result = default(T);
  277. try
  278. {
  279. result = function();
  280. }
  281. catch (Exception exception)
  282. {
  283. subject.OnError(exception);
  284. return;
  285. }
  286. subject.OnNext(result);
  287. subject.OnCompleted();
  288. });
  289. return subject.AsObservable();
  290. };
  291. }
  292. public static Func<IObservable<Unit>> ToAsync(Action action)
  293. {
  294. return ToAsync(action, Scheduler.DefaultSchedulers.AsyncConversions);
  295. }
  296. public static Func<IObservable<Unit>> ToAsync(Action action, IScheduler scheduler)
  297. {
  298. return () =>
  299. {
  300. var subject = new AsyncSubject<Unit>();
  301. scheduler.Schedule(() =>
  302. {
  303. try
  304. {
  305. action();
  306. }
  307. catch (Exception exception)
  308. {
  309. subject.OnError(exception);
  310. return;
  311. }
  312. subject.OnNext(Unit.Default);
  313. subject.OnCompleted();
  314. });
  315. return subject.AsObservable();
  316. };
  317. }
  318. }
  319. }