Observable.Single.cs 40 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708
  1. // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
  2. using System.Collections.Generic;
  3. using System.Reactive.Concurrency;
  4. namespace System.Reactive.Linq
  5. {
  6. public static partial class Observable
  7. {
  8. #region + AsObservable +
  9. /// <summary>
  10. /// Hides the identity of an observable sequence.
  11. /// </summary>
  12. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  13. /// <param name="source">An observable sequence whose identity to hide.</param>
  14. /// <returns>An observable sequence that hides the identity of the source sequence.</returns>
  15. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  16. public static IObservable<TSource> AsObservable<TSource>(this IObservable<TSource> source)
  17. {
  18. if (source == null)
  19. throw new ArgumentNullException("source");
  20. return s_impl.AsObservable<TSource>(source);
  21. }
  22. #endregion
  23. #region + Buffer +
  24. /// <summary>
  25. /// Projects each element of an observable sequence into consecutive non-overlapping buffers which are produced based on element count information.
  26. /// </summary>
  27. /// <typeparam name="TSource">The type of the elements in the source sequence, and in the lists in the result sequence.</typeparam>
  28. /// <param name="source">Source sequence to produce buffers over.</param>
  29. /// <param name="count">Length of each buffer.</param>
  30. /// <returns>An observable sequence of buffers.</returns>
  31. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  32. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> is less than or equal to zero.</exception>
  33. public static IObservable<IList<TSource>> Buffer<TSource>(this IObservable<TSource> source, int count)
  34. {
  35. if (source == null)
  36. throw new ArgumentNullException("source");
  37. if (count <= 0)
  38. throw new ArgumentOutOfRangeException("count");
  39. return s_impl.Buffer<TSource>(source, count);
  40. }
  41. /// <summary>
  42. /// Projects each element of an observable sequence into zero or more buffers which are produced based on element count information.
  43. /// </summary>
  44. /// <typeparam name="TSource">The type of the elements in the source sequence, and in the lists in the result sequence.</typeparam>
  45. /// <param name="source">Source sequence to produce buffers over.</param>
  46. /// <param name="count">Length of each buffer.</param>
  47. /// <param name="skip">Number of elements to skip between creation of consecutive buffers.</param>
  48. /// <returns>An observable sequence of buffers.</returns>
  49. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  50. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> or <paramref name="skip"/> is less than or equal to zero.</exception>
  51. public static IObservable<IList<TSource>> Buffer<TSource>(this IObservable<TSource> source, int count, int skip)
  52. {
  53. if (source == null)
  54. throw new ArgumentNullException("source");
  55. if (count <= 0)
  56. throw new ArgumentOutOfRangeException("count");
  57. if (skip <= 0)
  58. throw new ArgumentOutOfRangeException("skip");
  59. return s_impl.Buffer<TSource>(source, count, skip);
  60. }
  61. #endregion
  62. #region + Dematerialize +
  63. /// <summary>
  64. /// Dematerializes the explicit notification values of an observable sequence as implicit notifications.
  65. /// </summary>
  66. /// <typeparam name="TSource">The type of the elements materialized in the source sequence notification objects.</typeparam>
  67. /// <param name="source">An observable sequence containing explicit notification values which have to be turned into implicit notifications.</param>
  68. /// <returns>An observable sequence exhibiting the behavior corresponding to the source sequence's notification values.</returns>
  69. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  70. public static IObservable<TSource> Dematerialize<TSource>(this IObservable<Notification<TSource>> source)
  71. {
  72. if (source == null)
  73. throw new ArgumentNullException("source");
  74. return s_impl.Dematerialize<TSource>(source);
  75. }
  76. #endregion
  77. #region + DistinctUntilChanged +
  78. /// <summary>
  79. /// Returns an observable sequence that contains only distinct contiguous elements.
  80. /// </summary>
  81. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  82. /// <param name="source">An observable sequence to retain distinct contiguous elements for.</param>
  83. /// <returns>An observable sequence only containing the distinct contiguous elements from the source sequence.</returns>
  84. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  85. public static IObservable<TSource> DistinctUntilChanged<TSource>(this IObservable<TSource> source)
  86. {
  87. if (source == null)
  88. throw new ArgumentNullException("source");
  89. return s_impl.DistinctUntilChanged<TSource>(source);
  90. }
  91. /// <summary>
  92. /// Returns an observable sequence that contains only distinct contiguous elements according to the comparer.
  93. /// </summary>
  94. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  95. /// <param name="source">An observable sequence to retain distinct contiguous elements for.</param>
  96. /// <param name="comparer">Equality comparer for source elements.</param>
  97. /// <returns>An observable sequence only containing the distinct contiguous elements from the source sequence.</returns>
  98. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="comparer"/> is null.</exception>
  99. public static IObservable<TSource> DistinctUntilChanged<TSource>(this IObservable<TSource> source, IEqualityComparer<TSource> comparer)
  100. {
  101. if (source == null)
  102. throw new ArgumentNullException("source");
  103. if (comparer == null)
  104. throw new ArgumentNullException("comparer");
  105. return s_impl.DistinctUntilChanged<TSource>(source, comparer);
  106. }
  107. /// <summary>
  108. /// Returns an observable sequence that contains only distinct contiguous elements according to the keySelector.
  109. /// </summary>
  110. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  111. /// <typeparam name="TKey">The type of the discriminator key computed for each element in the source sequence.</typeparam>
  112. /// <param name="source">An observable sequence to retain distinct contiguous elements for, based on a computed key value.</param>
  113. /// <param name="keySelector">A function to compute the comparison key for each element.</param>
  114. /// <returns>An observable sequence only containing the distinct contiguous elements, based on a computed key value, from the source sequence.</returns>
  115. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="keySelector"/> is null.</exception>
  116. public static IObservable<TSource> DistinctUntilChanged<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector)
  117. {
  118. if (source == null)
  119. throw new ArgumentNullException("source");
  120. if (keySelector == null)
  121. throw new ArgumentNullException("keySelector");
  122. return s_impl.DistinctUntilChanged<TSource, TKey>(source, keySelector);
  123. }
  124. /// <summary>
  125. /// Returns an observable sequence that contains only distinct contiguous elements according to the keySelector and the comparer.
  126. /// </summary>
  127. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  128. /// <typeparam name="TKey">The type of the discriminator key computed for each element in the source sequence.</typeparam>
  129. /// <param name="source">An observable sequence to retain distinct contiguous elements for, based on a computed key value.</param>
  130. /// <param name="keySelector">A function to compute the comparison key for each element.</param>
  131. /// <param name="comparer">Equality comparer for computed key values.</param>
  132. /// <returns>An observable sequence only containing the distinct contiguous elements, based on a computed key value, from the source sequence.</returns>
  133. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="keySelector"/> or <paramref name="comparer"/> is null.</exception>
  134. public static IObservable<TSource> DistinctUntilChanged<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> comparer)
  135. {
  136. if (source == null)
  137. throw new ArgumentNullException("source");
  138. if (keySelector == null)
  139. throw new ArgumentNullException("keySelector");
  140. if (comparer == null)
  141. throw new ArgumentNullException("comparer");
  142. return s_impl.DistinctUntilChanged<TSource, TKey>(source, keySelector, comparer);
  143. }
  144. #endregion
  145. #region + Do +
  146. /// <summary>
  147. /// Invokes an action for each element in the observable sequence, and propagates all observer messages through the result sequence.
  148. /// This method can be used for debugging, logging, etc. of query behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline.
  149. /// </summary>
  150. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  151. /// <param name="source">Source sequence.</param>
  152. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  153. /// <returns>The source sequence with the side-effecting behavior applied.</returns>
  154. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> is null.</exception>
  155. public static IObservable<TSource> Do<TSource>(this IObservable<TSource> source, Action<TSource> onNext)
  156. {
  157. if (source == null)
  158. throw new ArgumentNullException("source");
  159. if (onNext == null)
  160. throw new ArgumentNullException("onNext");
  161. return s_impl.Do<TSource>(source, onNext);
  162. }
  163. /// <summary>
  164. /// Invokes an action for each element in the observable sequence and invokes an action upon graceful termination of the observable sequence.
  165. /// This method can be used for debugging, logging, etc. of query behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline.
  166. /// </summary>
  167. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  168. /// <param name="source">Source sequence.</param>
  169. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  170. /// <param name="onCompleted">Action to invoke upon graceful termination of the observable sequence.</param>
  171. /// <returns>The source sequence with the side-effecting behavior applied.</returns>
  172. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onCompleted"/> is null.</exception>
  173. public static IObservable<TSource> Do<TSource>(this IObservable<TSource> source, Action<TSource> onNext, Action onCompleted)
  174. {
  175. if (source == null)
  176. throw new ArgumentNullException("source");
  177. if (onNext == null)
  178. throw new ArgumentNullException("onNext");
  179. if (onCompleted == null)
  180. throw new ArgumentNullException("onCompleted");
  181. return s_impl.Do<TSource>(source, onNext, onCompleted);
  182. }
  183. /// <summary>
  184. /// Invokes an action for each element in the observable sequence and invokes an action upon exceptional termination of the observable sequence.
  185. /// This method can be used for debugging, logging, etc. of query behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline.
  186. /// </summary>
  187. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  188. /// <param name="source">Source sequence.</param>
  189. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  190. /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
  191. /// <returns>The source sequence with the side-effecting behavior applied.</returns>
  192. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onError"/> is null.</exception>
  193. public static IObservable<TSource> Do<TSource>(this IObservable<TSource> source, Action<TSource> onNext, Action<Exception> onError)
  194. {
  195. if (source == null)
  196. throw new ArgumentNullException("source");
  197. if (onNext == null)
  198. throw new ArgumentNullException("onNext");
  199. if (onError == null)
  200. throw new ArgumentNullException("onError");
  201. return s_impl.Do<TSource>(source, onNext, onError);
  202. }
  203. /// <summary>
  204. /// Invokes an action for each element in the observable sequence and invokes an action upon graceful or exceptional termination of the observable sequence.
  205. /// This method can be used for debugging, logging, etc. of query behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline.
  206. /// </summary>
  207. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  208. /// <param name="source">Source sequence.</param>
  209. /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
  210. /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
  211. /// <param name="onCompleted">Action to invoke upon graceful termination of the observable sequence.</param>
  212. /// <returns>The source sequence with the side-effecting behavior applied.</returns>
  213. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onError"/> or <paramref name="onCompleted"/> is null.</exception>
  214. public static IObservable<TSource> Do<TSource>(this IObservable<TSource> source, Action<TSource> onNext, Action<Exception> onError, Action onCompleted)
  215. {
  216. if (source == null)
  217. throw new ArgumentNullException("source");
  218. if (onNext == null)
  219. throw new ArgumentNullException("onNext");
  220. if (onError == null)
  221. throw new ArgumentNullException("onError");
  222. if (onCompleted == null)
  223. throw new ArgumentNullException("onCompleted");
  224. return s_impl.Do<TSource>(source, onNext, onError, onCompleted);
  225. }
  226. /// <summary>
  227. /// Invokes the observer's methods for each message in the source sequence.
  228. /// This method can be used for debugging, logging, etc. of query behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline.
  229. /// </summary>
  230. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  231. /// <param name="source">Source sequence.</param>
  232. /// <param name="observer">Observer whose methods to invoke as part of the source sequence's observation.</param>
  233. /// <returns>The source sequence with the side-effecting behavior applied.</returns>
  234. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="observer"/> is null.</exception>
  235. public static IObservable<TSource> Do<TSource>(this IObservable<TSource> source, IObserver<TSource> observer)
  236. {
  237. if (source == null)
  238. throw new ArgumentNullException("source");
  239. if (observer == null)
  240. throw new ArgumentNullException("observer");
  241. return s_impl.Do<TSource>(source, observer);
  242. }
  243. #endregion
  244. #region + Finally +
  245. /// <summary>
  246. /// Invokes a specified action after the source observable sequence terminates gracefully or exceptionally.
  247. /// </summary>
  248. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  249. /// <param name="source">Source sequence.</param>
  250. /// <param name="finallyAction">Action to invoke after the source observable sequence terminates.</param>
  251. /// <returns>Source sequence with the action-invoking termination behavior applied.</returns>
  252. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="finallyAction"/> is null.</exception>
  253. public static IObservable<TSource> Finally<TSource>(this IObservable<TSource> source, Action finallyAction)
  254. {
  255. if (source == null)
  256. throw new ArgumentNullException("source");
  257. if (finallyAction == null)
  258. throw new ArgumentNullException("finallyAction");
  259. return s_impl.Finally<TSource>(source, finallyAction);
  260. }
  261. #endregion
  262. #region + IgnoreElements +
  263. /// <summary>
  264. /// Ignores all elements in an observable sequence leaving only the termination messages.
  265. /// </summary>
  266. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  267. /// <param name="source">Source sequence.</param>
  268. /// <returns>An empty observable sequence that signals termination, successful or exceptional, of the source sequence.</returns>
  269. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  270. public static IObservable<TSource> IgnoreElements<TSource>(this IObservable<TSource> source)
  271. {
  272. if (source == null)
  273. throw new ArgumentNullException("source");
  274. return s_impl.IgnoreElements<TSource>(source);
  275. }
  276. #endregion
  277. #region + Materialize +
  278. /// <summary>
  279. /// Materializes the implicit notifications of an observable sequence as explicit notification values.
  280. /// </summary>
  281. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  282. /// <param name="source">An observable sequence to get notification values for.</param>
  283. /// <returns>An observable sequence containing the materialized notification values from the source sequence.</returns>
  284. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  285. public static IObservable<Notification<TSource>> Materialize<TSource>(this IObservable<TSource> source)
  286. {
  287. if (source == null)
  288. throw new ArgumentNullException("source");
  289. return s_impl.Materialize<TSource>(source);
  290. }
  291. #endregion
  292. #region + Repeat +
  293. /// <summary>
  294. /// Repeats the observable sequence indefinitely.
  295. /// </summary>
  296. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  297. /// <param name="source">Observable sequence to repeat.</param>
  298. /// <returns>The observable sequence producing the elements of the given sequence repeatedly and sequentially.</returns>
  299. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  300. public static IObservable<TSource> Repeat<TSource>(this IObservable<TSource> source)
  301. {
  302. if (source == null)
  303. throw new ArgumentNullException("source");
  304. return s_impl.Repeat<TSource>(source);
  305. }
  306. /// <summary>
  307. /// Repeats the observable sequence a specified number of times.
  308. /// </summary>
  309. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  310. /// <param name="source">Observable sequence to repeat.</param>
  311. /// <param name="repeatCount">Number of times to repeat the sequence.</param>
  312. /// <returns>The observable sequence producing the elements of the given sequence repeatedly.</returns>
  313. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  314. /// <exception cref="ArgumentOutOfRangeException"><paramref name="repeatCount"/> is less than zero.</exception>
  315. public static IObservable<TSource> Repeat<TSource>(this IObservable<TSource> source, int repeatCount)
  316. {
  317. if (source == null)
  318. throw new ArgumentNullException("source");
  319. if (repeatCount < 0)
  320. throw new ArgumentOutOfRangeException("repeatCount");
  321. return s_impl.Repeat<TSource>(source, repeatCount);
  322. }
  323. #endregion
  324. #region + Retry +
  325. /// <summary>
  326. /// Repeats the source observable sequence until it successfully terminates.
  327. /// </summary>
  328. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  329. /// <param name="source">Observable sequence to repeat until it successfully terminates.</param>
  330. /// <returns>An observable sequence producing the elements of the given sequence repeatedly until it terminates successfully.</returns>
  331. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  332. public static IObservable<TSource> Retry<TSource>(this IObservable<TSource> source)
  333. {
  334. if (source == null)
  335. throw new ArgumentNullException("source");
  336. return s_impl.Retry<TSource>(source);
  337. }
  338. /// <summary>
  339. /// Repeats the source observable sequence the specified number of times or until it successfully terminates.
  340. /// </summary>
  341. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  342. /// <param name="source">Observable sequence to repeat until it successfully terminates.</param>
  343. /// <param name="retryCount">Number of times to repeat the sequence.</param>
  344. /// <returns>An observable sequence producing the elements of the given sequence repeatedly until it terminates successfully.</returns>
  345. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  346. /// <exception cref="ArgumentOutOfRangeException"><paramref name="retryCount"/> is less than zero.</exception>
  347. public static IObservable<TSource> Retry<TSource>(this IObservable<TSource> source, int retryCount)
  348. {
  349. if (source == null)
  350. throw new ArgumentNullException("source");
  351. if (retryCount < 0)
  352. throw new ArgumentOutOfRangeException("retryCount");
  353. return s_impl.Retry<TSource>(source, retryCount);
  354. }
  355. #endregion
  356. #region + Scan +
  357. /// <summary>
  358. /// Applies an accumulator function over an observable sequence and returns each intermediate result. The specified seed value is used as the initial accumulator value.
  359. /// For aggregation behavior with no intermediate results, see <see cref="Observable.Aggregate&lt;TSource, Accumulate&gt;"/>.
  360. /// </summary>
  361. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  362. /// <typeparam name="TAccumulate">The type of the result of the aggregation.</typeparam>
  363. /// <param name="source">An observable sequence to accumulate over.</param>
  364. /// <param name="seed">The initial accumulator value.</param>
  365. /// <param name="accumulator">An accumulator function to be invoked on each element.</param>
  366. /// <returns>An observable sequence containing the accumulated values.</returns>
  367. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="accumulator"/> is null.</exception>
  368. public static IObservable<TAccumulate> Scan<TSource, TAccumulate>(this IObservable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> accumulator)
  369. {
  370. if (source == null)
  371. throw new ArgumentNullException("source");
  372. if (accumulator == null)
  373. throw new ArgumentNullException("accumulator");
  374. return s_impl.Scan<TSource, TAccumulate>(source, seed, accumulator);
  375. }
  376. /// <summary>
  377. /// Applies an accumulator function over an observable sequence and returns each intermediate result.
  378. /// For aggregation behavior with no intermediate results, see <see cref="Observable.Aggregate&lt;TSource&gt;"/>.
  379. /// </summary>
  380. /// <typeparam name="TSource">The type of the elements in the source sequence and the result of the aggregation.</typeparam>
  381. /// <param name="source">An observable sequence to accumulate over.</param>
  382. /// <param name="accumulator">An accumulator function to be invoked on each element.</param>
  383. /// <returns>An observable sequence containing the accumulated values.</returns>
  384. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="accumulator"/> is null.</exception>
  385. public static IObservable<TSource> Scan<TSource>(this IObservable<TSource> source, Func<TSource, TSource, TSource> accumulator)
  386. {
  387. if (source == null)
  388. throw new ArgumentNullException("source");
  389. if (accumulator == null)
  390. throw new ArgumentNullException("accumulator");
  391. return s_impl.Scan<TSource>(source, accumulator);
  392. }
  393. #endregion
  394. #region + SkipLast +
  395. /// <summary>
  396. /// Bypasses a specified number of elements at the end of an observable sequence.
  397. /// </summary>
  398. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  399. /// <param name="source">Source sequence.</param>
  400. /// <param name="count">Number of elements to bypass at the end of the source sequence.</param>
  401. /// <returns>An observable sequence containing the source sequence elements except for the bypassed ones at the end.</returns>
  402. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  403. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> is less than zero.</exception>
  404. /// <remarks>
  405. /// This operator accumulates a queue with a length enough to store the first <paramref name="count"/> elements. As more elements are
  406. /// received, elements are taken from the front of the queue and produced on the result sequence. This causes elements to be delayed.
  407. /// </remarks>
  408. public static IObservable<TSource> SkipLast<TSource>(this IObservable<TSource> source, int count)
  409. {
  410. if (source == null)
  411. throw new ArgumentNullException("source");
  412. if (count < 0)
  413. throw new ArgumentOutOfRangeException("count");
  414. return s_impl.SkipLast<TSource>(source, count);
  415. }
  416. #endregion
  417. #region + StartWith +
  418. /// <summary>
  419. /// Prepends a sequence of values to an observable sequence.
  420. /// </summary>
  421. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  422. /// <param name="source">Source sequence to prepend values to.</param>
  423. /// <param name="values">Values to prepend to the specified sequence.</param>
  424. /// <returns>The source sequence prepended with the specified values.</returns>
  425. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="values"/> is null.</exception>
  426. public static IObservable<TSource> StartWith<TSource>(this IObservable<TSource> source, params TSource[] values)
  427. {
  428. if (source == null)
  429. throw new ArgumentNullException("source");
  430. if (values == null)
  431. throw new ArgumentNullException("values");
  432. return s_impl.StartWith<TSource>(source, values);
  433. }
  434. /// <summary>
  435. /// Prepends a sequence of values to an observable sequence.
  436. /// </summary>
  437. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  438. /// <param name="source">Source sequence to prepend values to.</param>
  439. /// <param name="values">Values to prepend to the specified sequence.</param>
  440. /// <returns>The source sequence prepended with the specified values.</returns>
  441. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="values"/> is null.</exception>
  442. public static IObservable<TSource> StartWith<TSource>(this IObservable<TSource> source, IEnumerable<TSource> values)
  443. {
  444. if (source == null)
  445. throw new ArgumentNullException("source");
  446. if (values == null)
  447. throw new ArgumentNullException("values");
  448. TSource[] valueArray = values as TSource[];
  449. if (valueArray == null)
  450. {
  451. List<TSource> valueList = new List<TSource>(values);
  452. valueArray = valueList.ToArray();
  453. }
  454. return s_impl.StartWith<TSource>(source, valueArray);
  455. }
  456. /// <summary>
  457. /// Prepends a sequence of values to an observable sequence.
  458. /// </summary>
  459. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  460. /// <param name="source">Source sequence to prepend values to.</param>
  461. /// <param name="scheduler">Scheduler to emit the prepended values on.</param>
  462. /// <param name="values">Values to prepend to the specified sequence.</param>
  463. /// <returns>The source sequence prepended with the specified values.</returns>
  464. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="scheduler"/> or <paramref name="values"/> is null.</exception>
  465. public static IObservable<TSource> StartWith<TSource>(this IObservable<TSource> source, IScheduler scheduler, params TSource[] values)
  466. {
  467. if (source == null)
  468. throw new ArgumentNullException("source");
  469. if (scheduler == null)
  470. throw new ArgumentNullException("scheduler");
  471. if (values == null)
  472. throw new ArgumentNullException("values");
  473. return s_impl.StartWith<TSource>(source, scheduler, values);
  474. }
  475. /// <summary>
  476. /// Prepends a sequence of values to an observable sequence.
  477. /// </summary>
  478. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  479. /// <param name="source">Source sequence to prepend values to.</param>
  480. /// <param name="scheduler">Scheduler to emit the prepended values on.</param>
  481. /// <param name="values">Values to prepend to the specified sequence.</param>
  482. /// <returns>The source sequence prepended with the specified values.</returns>
  483. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="scheduler"/> or <paramref name="values"/> is null.</exception>
  484. public static IObservable<TSource> StartWith<TSource>(this IObservable<TSource> source, IScheduler scheduler, IEnumerable<TSource> values)
  485. {
  486. if (source == null)
  487. throw new ArgumentNullException("source");
  488. if (scheduler == null)
  489. throw new ArgumentNullException("scheduler");
  490. if (values == null)
  491. throw new ArgumentNullException("values");
  492. TSource[] valueArray = values as TSource[];
  493. if (valueArray == null)
  494. {
  495. List<TSource> valueList = new List<TSource>(values);
  496. valueArray = valueList.ToArray();
  497. }
  498. return s_impl.StartWith<TSource>(source, scheduler, valueArray);
  499. }
  500. #endregion
  501. #region + TakeLast +
  502. /// <summary>
  503. /// Returns a specified number of contiguous elements from the end of an observable sequence.
  504. /// </summary>
  505. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  506. /// <param name="source">Source sequence.</param>
  507. /// <param name="count">Number of elements to take from the end of the source sequence.</param>
  508. /// <returns>An observable sequence containing the specified number of elements from the end of the source sequence.</returns>
  509. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  510. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> is less than zero.</exception>
  511. /// <remarks>
  512. /// This operator accumulates a buffer with a length enough to store elements <paramref name="count"/> elements. Upon completion of
  513. /// the source sequence, this buffer is drained on the result sequence. This causes the elements to be delayed.
  514. /// </remarks>
  515. public static IObservable<TSource> TakeLast<TSource>(this IObservable<TSource> source, int count)
  516. {
  517. if (source == null)
  518. throw new ArgumentNullException("source");
  519. if (count < 0)
  520. throw new ArgumentOutOfRangeException("count");
  521. return s_impl.TakeLast<TSource>(source, count);
  522. }
  523. /// <summary>
  524. /// Returns a specified number of contiguous elements from the end of an observable sequence, using the specified scheduler to drain the queue.
  525. /// </summary>
  526. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  527. /// <param name="source">Source sequence.</param>
  528. /// <param name="count">Number of elements to take from the end of the source sequence.</param>
  529. /// <param name="scheduler">Scheduler used to drain the queue upon completion of the source sequence.</param>
  530. /// <returns>An observable sequence containing the specified number of elements from the end of the source sequence.</returns>
  531. /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="scheduler"/> is null.</exception>
  532. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> is less than zero.</exception>
  533. /// <remarks>
  534. /// This operator accumulates a buffer with a length enough to store elements <paramref name="count"/> elements. Upon completion of
  535. /// the source sequence, this buffer is drained on the result sequence. This causes the elements to be delayed.
  536. /// </remarks>
  537. public static IObservable<TSource> TakeLast<TSource>(this IObservable<TSource> source, int count, IScheduler scheduler)
  538. {
  539. if (source == null)
  540. throw new ArgumentNullException("source");
  541. if (count < 0)
  542. throw new ArgumentOutOfRangeException("count");
  543. if (scheduler == null)
  544. throw new ArgumentNullException("scheduler");
  545. return s_impl.TakeLast<TSource>(source, count, scheduler);
  546. }
  547. #endregion
  548. #region + TakeLastBuffer +
  549. /// <summary>
  550. /// Returns a list with the specified number of contiguous elements from the end of an observable sequence.
  551. /// </summary>
  552. /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
  553. /// <param name="source">Source sequence.</param>
  554. /// <param name="count">Number of elements to take from the end of the source sequence.</param>
  555. /// <returns>An observable sequence containing a single list with the specified number of elements from the end of the source sequence.</returns>
  556. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  557. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> is less than zero.</exception>
  558. /// <remarks>
  559. /// This operator accumulates a buffer with a length enough to store <paramref name="count"/> elements. Upon completion of the
  560. /// source sequence, this buffer is produced on the result sequence.
  561. /// </remarks>
  562. public static IObservable<IList<TSource>> TakeLastBuffer<TSource>(this IObservable<TSource> source, int count)
  563. {
  564. if (source == null)
  565. throw new ArgumentNullException("source");
  566. if (count < 0)
  567. throw new ArgumentOutOfRangeException("count");
  568. return s_impl.TakeLastBuffer<TSource>(source, count);
  569. }
  570. #endregion
  571. #region + Window +
  572. /// <summary>
  573. /// Projects each element of an observable sequence into consecutive non-overlapping windows which are produced based on element count information.
  574. /// </summary>
  575. /// <typeparam name="TSource">The type of the elements in the source sequence, and in the windows in the result sequence.</typeparam>
  576. /// <param name="source">Source sequence to produce windows over.</param>
  577. /// <param name="count">Length of each window.</param>
  578. /// <returns>An observable sequence of windows.</returns>
  579. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  580. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> is less than or equal to zero.</exception>
  581. public static IObservable<IObservable<TSource>> Window<TSource>(this IObservable<TSource> source, int count)
  582. {
  583. if (source == null)
  584. throw new ArgumentNullException("source");
  585. if (count <= 0)
  586. throw new ArgumentOutOfRangeException("count");
  587. return s_impl.Window<TSource>(source, count);
  588. }
  589. /// <summary>
  590. /// Projects each element of an observable sequence into zero or more windows which are produced based on element count information.
  591. /// </summary>
  592. /// <typeparam name="TSource">The type of the elements in the source sequence, and in the windows in the result sequence.</typeparam>
  593. /// <param name="source">Source sequence to produce windows over.</param>
  594. /// <param name="count">Length of each window.</param>
  595. /// <param name="skip">Number of elements to skip between creation of consecutive windows.</param>
  596. /// <returns>An observable sequence of windows.</returns>
  597. /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
  598. /// <exception cref="ArgumentOutOfRangeException"><paramref name="count"/> or <paramref name="skip"/> is less than or equal to zero.</exception>
  599. public static IObservable<IObservable<TSource>> Window<TSource>(this IObservable<TSource> source, int count, int skip)
  600. {
  601. if (source == null)
  602. throw new ArgumentNullException("source");
  603. if (count <= 0)
  604. throw new ArgumentOutOfRangeException("count");
  605. if (skip <= 0)
  606. throw new ArgumentOutOfRangeException("skip");
  607. return s_impl.Window<TSource>(source, count, skip);
  608. }
  609. #endregion
  610. }
  611. }