// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
using System.Collections.Generic;
using System.Reactive.Concurrency;
namespace System.Reactive.Linq
{
public static partial class Observable
{
#region + Buffer +
#region TimeSpan only
///
/// Projects each element of an observable sequence into consecutive non-overlapping buffers which are produced based on timing information.
///
/// The type of the elements in the source sequence, and in the lists in the result sequence.
/// Source sequence to produce buffers over.
/// Length of each buffer.
/// An observable sequence of buffers.
/// is null.
/// is less than TimeSpan.Zero.
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers as fast as it can.
/// Because all source sequence elements end up in one of the buffers, some buffers won't have a zero time span. This is a side-effect of the asynchrony introduced
/// by the scheduler, where the action to close the current buffer and to create a new buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
public static IObservable> Buffer(this IObservable source, TimeSpan timeSpan)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeSpan < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeSpan");
return s_impl.Buffer(source, timeSpan);
}
///
/// Projects each element of an observable sequence into consecutive non-overlapping buffers which are produced based on timing information, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence, and in the lists in the result sequence.
/// Source sequence to produce buffers over.
/// Length of each buffer.
/// Scheduler to run buffering timers on.
/// An observable sequence of buffers.
/// or is null.
/// is less than TimeSpan.Zero.
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers as fast as it can.
/// Because all source sequence elements end up in one of the buffers, some buffers won't have a zero time span. This is a side-effect of the asynchrony introduced
/// by the scheduler, where the action to close the current buffer and to create a new buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
public static IObservable> Buffer(this IObservable source, TimeSpan timeSpan, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeSpan < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeSpan");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Buffer(source, timeSpan, scheduler);
}
///
/// Projects each element of an observable sequence into zero or more buffers which are produced based on timing information.
///
/// The type of the elements in the source sequence, and in the lists in the result sequence.
/// Source sequence to produce buffers over.
/// Length of each buffer.
/// Interval between creation of consecutive buffers.
/// An observable sequence of buffers.
/// is null.
/// or is less than TimeSpan.Zero.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers with minimum duration
/// length. However, some buffers won't have a zero time span. This is a side-effect of the asynchrony introduced by the scheduler, where the action to close the
/// current buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers as fast as it can.
/// However, this doesn't mean all buffers will start at the beginning of the source sequence. This is a side-effect of the asynchrony introduced by the scheduler,
/// where the action to create a new buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
///
public static IObservable> Buffer(this IObservable source, TimeSpan timeSpan, TimeSpan timeShift)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeSpan < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeSpan");
if (timeShift < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeShift");
return s_impl.Buffer(source, timeSpan, timeShift);
}
///
/// Projects each element of an observable sequence into zero or more buffers which are produced based on timing information, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence, and in the lists in the result sequence.
/// Source sequence to produce buffers over.
/// Length of each buffer.
/// Interval between creation of consecutive buffers.
/// Scheduler to run buffering timers on.
/// An observable sequence of buffers.
/// or is null.
/// or is less than TimeSpan.Zero.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers with minimum duration
/// length. However, some buffers won't have a zero time span. This is a side-effect of the asynchrony introduced by the scheduler, where the action to close the
/// current buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers as fast as it can.
/// However, this doesn't mean all buffers will start at the beginning of the source sequence. This is a side-effect of the asynchrony introduced by the scheduler,
/// where the action to create a new buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
///
public static IObservable> Buffer(this IObservable source, TimeSpan timeSpan, TimeSpan timeShift, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeSpan < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeSpan");
if (timeShift < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeShift");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Buffer(source, timeSpan, timeShift, scheduler);
}
#endregion
#region TimeSpan + int
///
/// Projects each element of an observable sequence into a buffer that's sent out when either it's full or a given amount of time has elapsed.
/// A useful real-world analogy of this overload is the behavior of a ferry leaving the dock when all seats are taken, or at the scheduled time of departure, whichever event occurs first.
///
/// The type of the elements in the source sequence, and in the lists in the result sequence.
/// Source sequence to produce buffers over.
/// Maximum time length of a window.
/// Maximum element count of a window.
/// An observable sequence of buffers.
/// is null.
/// is less than TimeSpan.Zero. -or- is less than or equal to zero.
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers as fast as it can.
/// Because all source sequence elements end up in one of the buffers, some buffers won't have a zero time span. This is a side-effect of the asynchrony introduced
/// by the scheduler, where the action to close the current buffer and to create a new buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
public static IObservable> Buffer(this IObservable source, TimeSpan timeSpan, int count)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeSpan < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeSpan");
if (count <= 0)
throw new ArgumentOutOfRangeException("count");
return s_impl.Buffer(source, timeSpan, count);
}
///
/// Projects each element of an observable sequence into a buffer that's sent out when either it's full or a given amount of time has elapsed, using the specified scheduler to run timers.
/// A useful real-world analogy of this overload is the behavior of a ferry leaving the dock when all seats are taken, or at the scheduled time of departure, whichever event occurs first.
///
/// The type of the elements in the source sequence, and in the lists in the result sequence.
/// Source sequence to produce buffers over.
/// Maximum time length of a buffer.
/// Maximum element count of a buffer.
/// Scheduler to run buffering timers on.
/// An observable sequence of buffers.
/// or is null.
/// is less than TimeSpan.Zero. -or- is less than or equal to zero.
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing the scheduler to create buffers as fast as it can.
/// Because all source sequence elements end up in one of the buffers, some buffers won't have a zero time span. This is a side-effect of the asynchrony introduced
/// by the scheduler, where the action to close the current buffer and to create a new buffer may not execute immediately, despite the TimeSpan.Zero due time.
///
public static IObservable> Buffer(this IObservable source, TimeSpan timeSpan, int count, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeSpan < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("timeSpan");
if (count <= 0)
throw new ArgumentOutOfRangeException("count");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Buffer(source, timeSpan, count, scheduler);
}
#endregion
#endregion
#region + Delay +
#region TimeSpan
///
/// Time shifts the observable sequence by the specified relative time duration.
/// The relative time intervals between the values are preserved.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay values for.
/// Relative time by which to shift the observable sequence. If this value is equal to TimeSpan.Zero, the scheduler will dispatch observer callbacks as soon as possible.
/// Time-shifted sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
///
/// This operator is less efficient than DelaySubscription because it records all notifications and time-delays those. This allows for immediate propagation of errors.
///
///
/// Observer callbacks for the resulting sequence will be run on the default scheduler. This effect is similar to using ObserveOn.
///
///
/// Exceptions signaled by the source sequence through an OnError callback are forwarded immediately to the result sequence. Any OnNext notifications that were in the queue at the point of the OnError callback will be dropped.
/// In order to delay error propagation, consider using the Observable.Materialize and Observable.Dematerialize operators, or use DelaySubscription.
///
///
public static IObservable Delay(this IObservable source, TimeSpan dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
return s_impl.Delay(source, dueTime);
}
///
/// Time shifts the observable sequence by the specified relative time duration, using the specified scheduler to run timers.
/// The relative time intervals between the values are preserved.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay values for.
/// Relative time by which to shift the observable sequence. If this value is equal to TimeSpan.Zero, the scheduler will dispatch observer callbacks as soon as possible.
/// Scheduler to run the delay timers on.
/// Time-shifted sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
///
/// This operator is less efficient than DelaySubscription because it records all notifications and time-delays those. This allows for immediate propagation of errors.
///
///
/// Observer callbacks for the resulting sequence will be run on the specified scheduler. This effect is similar to using ObserveOn.
///
///
/// Exceptions signaled by the source sequence through an OnError callback are forwarded immediately to the result sequence. Any OnNext notifications that were in the queue at the point of the OnError callback will be dropped.
///
///
/// Exceptions signaled by the source sequence through an OnError callback are forwarded immediately to the result sequence. Any OnNext notifications that were in the queue at the point of the OnError callback will be dropped.
/// In order to delay error propagation, consider using the Observable.Materialize and Observable.Dematerialize operators, or use DelaySubscription.
///
///
public static IObservable Delay(this IObservable source, TimeSpan dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Delay(source, dueTime, scheduler);
}
#endregion
#region DateTimeOffset
///
/// Time shifts the observable sequence to start propagating notifications at the specified absolute time.
/// The relative time intervals between the values are preserved.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay values for.
/// Absolute time used to shift the observable sequence; the relative time shift gets computed upon subscription. If this value is less than or equal to DateTimeOffset.UtcNow, the scheduler will dispatch observer callbacks as soon as possible.
/// Time-shifted sequence.
/// is null.
///
///
/// This operator is less efficient than DelaySubscription because it records all notifications and time-delays those. This allows for immediate propagation of errors.
///
///
/// Observer callbacks for the resulting sequence will be run on the default scheduler. This effect is similar to using ObserveOn.
///
///
/// Exceptions signaled by the source sequence through an OnError callback are forwarded immediately to the result sequence. Any OnNext notifications that were in the queue at the point of the OnError callback will be dropped.
/// In order to delay error propagation, consider using the Observable.Materialize and Observable.Dematerialize operators, or use DelaySubscription.
///
///
public static IObservable Delay(this IObservable source, DateTimeOffset dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
return s_impl.Delay(source, dueTime);
}
///
/// Time shifts the observable sequence to start propagating notifications at the specified absolute time, using the specified scheduler to run timers.
/// The relative time intervals between the values are preserved.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay values for.
/// Absolute time used to shift the observable sequence; the relative time shift gets computed upon subscription. If this value is less than or equal to DateTimeOffset.UtcNow, the scheduler will dispatch observer callbacks as soon as possible.
/// Scheduler to run the delay timers on.
/// Time-shifted sequence.
/// or is null.
///
///
/// This operator is less efficient than DelaySubscription because it records all notifications and time-delays those. This allows for immediate propagation of errors.
///
///
/// Observer callbacks for the resulting sequence will be run on the specified scheduler. This effect is similar to using ObserveOn.
///
///
/// Exceptions signaled by the source sequence through an OnError callback are forwarded immediately to the result sequence. Any OnNext notifications that were in the queue at the point of the OnError callback will be dropped.
/// In order to delay error propagation, consider using the Observable.Materialize and Observable.Dematerialize operators, or use DelaySubscription.
///
///
public static IObservable Delay(this IObservable source, DateTimeOffset dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Delay(source, dueTime, scheduler);
}
#endregion
#region Duration selector
///
/// Time shifts the observable sequence based on a delay selector function for each element.
///
/// The type of the elements in the source sequence.
/// The type of the elements in the delay sequences used to denote the delay duration of each element in the source sequence.
/// Source sequence to delay values for.
/// Selector function to retrieve a sequence indicating the delay for each given element.
/// Time-shifted sequence.
/// or is null.
public static IObservable Delay(this IObservable source, Func> delayDurationSelector)
{
if (source == null)
throw new ArgumentNullException("source");
if (delayDurationSelector == null)
throw new ArgumentNullException("delayDurationSelector");
return s_impl.Delay(source, delayDurationSelector);
}
///
/// Time shifts the observable sequence based on a subscription delay and a delay selector function for each element.
///
/// The type of the elements in the source sequence.
/// The type of the elements in the delay sequences used to denote the delay duration of each element in the source sequence.
/// Source sequence to delay values for.
/// Sequence indicating the delay for the subscription to the source.
/// Selector function to retrieve a sequence indicating the delay for each given element.
/// Time-shifted sequence.
/// or or is null.
public static IObservable Delay(this IObservable source, IObservable subscriptionDelay, Func> delayDurationSelector)
{
if (source == null)
throw new ArgumentNullException("source");
if (subscriptionDelay == null)
throw new ArgumentNullException("subscriptionDelay");
if (delayDurationSelector == null)
throw new ArgumentNullException("delayDurationSelector");
return s_impl.Delay(source, subscriptionDelay, delayDurationSelector);
}
#endregion
#endregion
#region + DelaySubscription +
///
/// Time shifts the observable sequence by delaying the subscription with the specified relative time duration.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay subscription for.
/// Relative time shift of the subscription.
/// Time-shifted sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
///
/// This operator is more efficient than Delay but postpones all side-effects of subscription and affects error propagation timing.
///
///
/// The side-effects of subscribing to the source sequence will be run on the default scheduler. Observer callbacks will not be affected.
///
///
public static IObservable DelaySubscription(this IObservable source, TimeSpan dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
return s_impl.DelaySubscription(source, dueTime);
}
///
/// Time shifts the observable sequence by delaying the subscription with the specified relative time duration, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay subscription for.
/// Relative time shift of the subscription.
/// Scheduler to run the subscription delay timer on.
/// Time-shifted sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
///
/// This operator is more efficient than Delay but postpones all side-effects of subscription and affects error propagation timing.
///
///
/// The side-effects of subscribing to the source sequence will be run on the specified scheduler. Observer callbacks will not be affected.
///
///
public static IObservable DelaySubscription(this IObservable source, TimeSpan dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.DelaySubscription(source, dueTime, scheduler);
}
///
/// Time shifts the observable sequence by delaying the subscription to the specified absolute time.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay subscription for.
/// Absolute time to perform the subscription at.
/// Time-shifted sequence.
/// is null.
///
///
/// This operator is more efficient than Delay but postpones all side-effects of subscription and affects error propagation timing.
///
///
/// The side-effects of subscribing to the source sequence will be run on the default scheduler. Observer callbacks will not be affected.
///
///
public static IObservable DelaySubscription(this IObservable source, DateTimeOffset dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
return s_impl.DelaySubscription(source, dueTime);
}
///
/// Time shifts the observable sequence by delaying the subscription to the specified absolute time, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to delay subscription for.
/// Absolute time to perform the subscription at.
/// Scheduler to run the subscription delay timer on.
/// Time-shifted sequence.
/// or is null.
///
///
/// This operator is more efficient than Delay but postpones all side-effects of subscription and affects error propagation timing.
///
///
/// The side-effects of subscribing to the source sequence will be run on the specified scheduler. Observer callbacks will not be affected.
///
///
public static IObservable DelaySubscription(this IObservable source, DateTimeOffset dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.DelaySubscription(source, dueTime, scheduler);
}
#endregion
#region + Generate +
///
/// Generates an observable sequence by running a state-driven and temporal loop producing the sequence's elements.
///
/// The type of the state used in the generator loop.
/// The type of the elements in the produced sequence.
/// Initial state.
/// Condition to terminate generation (upon returning false).
/// Iteration step function.
/// Selector function for results produced in the sequence.
/// Time selector function to control the speed of values being produced each iteration.
/// The generated sequence.
/// or or or is null.
public static IObservable Generate(TState initialState, Func condition, Func iterate, Func resultSelector, Func timeSelector)
{
if (condition == null)
throw new ArgumentNullException("condition");
if (iterate == null)
throw new ArgumentNullException("iterate");
if (resultSelector == null)
throw new ArgumentNullException("resultSelector");
if (timeSelector == null)
throw new ArgumentNullException("timeSelector");
return s_impl.Generate(initialState, condition, iterate, resultSelector, timeSelector);
}
///
/// Generates an observable sequence by running a state-driven and temporal loop producing the sequence's elements, using the specified scheduler to run timers and to send out observer messages.
///
/// The type of the state used in the generator loop.
/// The type of the elements in the produced sequence.
/// Initial state.
/// Condition to terminate generation (upon returning false).
/// Iteration step function.
/// Selector function for results produced in the sequence.
/// Time selector function to control the speed of values being produced each iteration.
/// Scheduler on which to run the generator loop.
/// The generated sequence.
/// or or or or is null.
public static IObservable Generate(TState initialState, Func condition, Func iterate, Func resultSelector, Func timeSelector, IScheduler scheduler)
{
if (condition == null)
throw new ArgumentNullException("condition");
if (iterate == null)
throw new ArgumentNullException("iterate");
if (resultSelector == null)
throw new ArgumentNullException("resultSelector");
if (timeSelector == null)
throw new ArgumentNullException("timeSelector");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Generate(initialState, condition, iterate, resultSelector, timeSelector, scheduler);
}
///
/// Generates an observable sequence by running a state-driven and temporal loop producing the sequence's elements.
///
/// The type of the state used in the generator loop.
/// The type of the elements in the produced sequence.
/// Initial state.
/// Condition to terminate generation (upon returning false).
/// Iteration step function.
/// Selector function for results produced in the sequence.
/// Time selector function to control the speed of values being produced each iteration.
/// The generated sequence.
/// or or or is null.
public static IObservable Generate(TState initialState, Func condition, Func iterate, Func resultSelector, Func timeSelector)
{
if (condition == null)
throw new ArgumentNullException("condition");
if (iterate == null)
throw new ArgumentNullException("iterate");
if (resultSelector == null)
throw new ArgumentNullException("resultSelector");
if (timeSelector == null)
throw new ArgumentNullException("timeSelector");
return s_impl.Generate(initialState, condition, iterate, resultSelector, timeSelector);
}
///
/// Generates an observable sequence by running a state-driven and temporal loop producing the sequence's elements, using the specified scheduler to run timers and to send out observer messages.
///
/// The type of the state used in the generator loop.
/// The type of the elements in the produced sequence.
/// Initial state.
/// Condition to terminate generation (upon returning false).
/// Iteration step function.
/// Selector function for results produced in the sequence.
/// Time selector function to control the speed of values being produced each iteration.
/// Scheduler on which to run the generator loop.
/// The generated sequence.
/// or or or or is null.
public static IObservable Generate(TState initialState, Func condition, Func iterate, Func resultSelector, Func timeSelector, IScheduler scheduler)
{
if (condition == null)
throw new ArgumentNullException("condition");
if (iterate == null)
throw new ArgumentNullException("iterate");
if (resultSelector == null)
throw new ArgumentNullException("resultSelector");
if (timeSelector == null)
throw new ArgumentNullException("timeSelector");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Generate(initialState, condition, iterate, resultSelector, timeSelector, scheduler);
}
#endregion
#region + Interval +
///
/// Returns an observable sequence that produces a value after each period.
///
/// Period for producing the values in the resulting sequence. If this value is equal to TimeSpan.Zero, the timer will recur as fast as possible.
/// An observable sequence that produces a value after each period.
/// is less than TimeSpan.Zero.
///
/// Intervals are measured between the start of subsequent notifications, not between the end of the previous and the start of the next notification.
/// If the observer takes longer than the interval period to handle the message, the subsequent notification will be delivered immediately after the
/// current one has been handled. In case you need to control the time between the end and the start of consecutive notifications, consider using the
///
/// operator instead.
///
public static IObservable Interval(TimeSpan period)
{
if (period < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("period");
return s_impl.Interval(period);
}
///
/// Returns an observable sequence that produces a value after each period, using the specified scheduler to run timers and to send out observer messages.
///
/// Period for producing the values in the resulting sequence. If this value is equal to TimeSpan.Zero, the timer will recur as fast as possible.
/// Scheduler to run the timer on.
/// An observable sequence that produces a value after each period.
/// is less than TimeSpan.Zero.
/// is null.
///
/// Intervals are measured between the start of subsequent notifications, not between the end of the previous and the start of the next notification.
/// If the observer takes longer than the interval period to handle the message, the subsequent notification will be delivered immediately after the
/// current one has been handled. In case you need to control the time between the end and the start of consecutive notifications, consider using the
///
/// operator instead.
///
public static IObservable Interval(TimeSpan period, IScheduler scheduler)
{
if (period < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("period");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Interval(period, scheduler);
}
#endregion
#region + Sample +
///
/// Samples the observable sequence at each interval.
/// Upon each sampling tick, the latest element (if any) in the source sequence during the last sampling interval is sent to the resulting sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to sample.
/// Interval at which to sample. If this value is equal to TimeSpan.Zero, the scheduler will continuously sample the stream.
/// Sampled observable sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
/// Specifying a TimeSpan.Zero value for doesn't guarantee all source sequence elements will be preserved. This is a side-effect
/// of the asynchrony introduced by the scheduler, where the sampling action may not execute immediately, despite the TimeSpan.Zero due time.
///
public static IObservable Sample(this IObservable source, TimeSpan interval)
{
if (source == null)
throw new ArgumentNullException("source");
if (interval < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("interval");
return s_impl.Sample(source, interval);
}
///
/// Samples the observable sequence at each interval, using the specified scheduler to run sampling timers.
/// Upon each sampling tick, the latest element (if any) in the source sequence during the last sampling interval is sent to the resulting sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to sample.
/// Interval at which to sample. If this value is equal to TimeSpan.Zero, the scheduler will continuously sample the stream.
/// Scheduler to run the sampling timer on.
/// Sampled observable sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
/// Specifying a TimeSpan.Zero value for doesn't guarantee all source sequence elements will be preserved. This is a side-effect
/// of the asynchrony introduced by the scheduler, where the sampling action may not execute immediately, despite the TimeSpan.Zero due time.
///
public static IObservable Sample(this IObservable source, TimeSpan interval, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (interval < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("interval");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Sample(source, interval, scheduler);
}
///
/// Samples the source observable sequence using a samper observable sequence producing sampling ticks.
/// Upon each sampling tick, the latest element (if any) in the source sequence during the last sampling interval is sent to the resulting sequence.
///
/// The type of the elements in the source sequence.
/// The type of the elements in the sampling sequence.
/// Source sequence to sample.
/// Sampling tick sequence.
/// Sampled observable sequence.
/// or is null.
public static IObservable Sample(this IObservable source, IObservable sampler)
{
if (source == null)
throw new ArgumentNullException("source");
if (sampler == null)
throw new ArgumentNullException("sampler");
return s_impl.Sample(source, sampler);
}
#endregion
#region + Skip +
///
/// Skips elements for the specified duration from the start of the observable source sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to skip elements for.
/// Duration for skipping elements from the start of the sequence.
/// An observable sequence with the elements skipped during the specified duration from the start of the source sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
///
/// Specifying a TimeSpan.Zero value for doesn't guarantee no elements will be dropped from the start of the source sequence.
/// This is a side-effect of the asynchrony introduced by the scheduler, where the action that causes callbacks from the source sequence to be forwarded
/// may not execute immediately, despite the TimeSpan.Zero due time.
///
///
/// Errors produced by the source sequence are always forwarded to the result sequence, even if the error occurs before the .
///
///
public static IObservable Skip(this IObservable source, TimeSpan duration)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
return s_impl.Skip(source, duration);
}
///
/// Skips elements for the specified duration from the start of the observable source sequence, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to skip elements for.
/// Duration for skipping elements from the start of the sequence.
/// Scheduler to run the timer on.
/// An observable sequence with the elements skipped during the specified duration from the start of the source sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
///
/// Specifying a TimeSpan.Zero value for doesn't guarantee no elements will be dropped from the start of the source sequence.
/// This is a side-effect of the asynchrony introduced by the scheduler, where the action that causes callbacks from the source sequence to be forwarded
/// may not execute immediately, despite the TimeSpan.Zero due time.
///
///
/// Errors produced by the source sequence are always forwarded to the result sequence, even if the error occurs before the .
///
///
public static IObservable Skip(this IObservable source, TimeSpan duration, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Skip(source, duration, scheduler);
}
#endregion
#region + SkipLast +
///
/// Skips elements for the specified duration from the end of the observable source sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to skip elements for.
/// Duration for skipping elements from the end of the sequence.
/// An observable sequence with the elements skipped during the specified duration from the end of the source sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a queue with a length enough to store elements received during the initial window.
/// As more elements are received, elements older than the specified are taken from the queue and produced on the
/// result sequence. This causes elements to be delayed with .
///
public static IObservable SkipLast(this IObservable source, TimeSpan duration)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
return s_impl.SkipLast(source, duration);
}
///
/// Skips elements for the specified duration from the end of the observable source sequence, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to skip elements for.
/// Duration for skipping elements from the end of the sequence.
/// Scheduler to run the timer on.
/// An observable sequence with the elements skipped during the specified duration from the end of the source sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a queue with a length enough to store elements received during the initial window.
/// As more elements are received, elements older than the specified are taken from the queue and produced on the
/// result sequence. This causes elements to be delayed with .
///
public static IObservable SkipLast(this IObservable source, TimeSpan duration, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.SkipLast(source, duration, scheduler);
}
#endregion
#region + SkipUntil +
///
/// Skips elements from the observable source sequence until the specified start time.
///
/// The type of the elements in the source sequence.
/// Source sequence to skip elements for.
/// Time to start taking elements from the source sequence. If this value is less than or equal to DateTimeOffset.UtcNow, no elements will be skipped.
/// An observable sequence with the elements skipped until the specified start time.
/// is null.
///
/// Errors produced by the source sequence are always forwarded to the result sequence, even if the error occurs before the .
///
public static IObservable SkipUntil(this IObservable source, DateTimeOffset startTime)
{
if (source == null)
throw new ArgumentNullException("source");
return s_impl.SkipUntil(source, startTime);
}
///
/// Skips elements from the observable source sequence until the specified start time, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to skip elements for.
/// Time to start taking elements from the source sequence. If this value is less than or equal to DateTimeOffset.UtcNow, no elements will be skipped.
/// Scheduler to run the timer on.
/// An observable sequence with the elements skipped until the specified start time.
/// or is null.
///
/// Errors produced by the source sequence are always forwarded to the result sequence, even if the error occurs before the .
///
public static IObservable SkipUntil(this IObservable source, DateTimeOffset startTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.SkipUntil(source, startTime, scheduler);
}
#endregion
#region + Take +
///
/// Takes elements for the specified duration from the start of the observable source sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the start of the sequence.
/// An observable sequence with the elements taken during the specified duration from the start of the source sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
/// Specifying a TimeSpan.Zero value for doesn't guarantee an empty sequence will be returned. This is a side-effect
/// of the asynchrony introduced by the scheduler, where the action that stops forwarding callbacks from the source sequence may not execute
/// immediately, despite the TimeSpan.Zero due time.
///
public static IObservable Take(this IObservable source, TimeSpan duration)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
return s_impl.Take(source, duration);
}
///
/// Takes elements for the specified duration from the start of the observable source sequence, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the start of the sequence.
/// Scheduler to run the timer on.
/// An observable sequence with the elements taken during the specified duration from the start of the source sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
/// Specifying a TimeSpan.Zero value for doesn't guarantee an empty sequence will be returned. This is a side-effect
/// of the asynchrony introduced by the scheduler, where the action that stops forwarding callbacks from the source sequence may not execute
/// immediately, despite the TimeSpan.Zero due time.
///
public static IObservable Take(this IObservable source, TimeSpan duration, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Take(source, duration, scheduler);
}
#endregion
#region + TakeLast +
///
/// Returns elements within the specified duration from the end of the observable source sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the end of the sequence.
/// An observable sequence with the elements taken during the specified duration from the end of the source sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a buffer with a length enough to store elements for any window during the lifetime of
/// the source sequence. Upon completion of the source sequence, this buffer is drained on the result sequence. This causes the result elements
/// to be delayed with .
///
public static IObservable TakeLast(this IObservable source, TimeSpan duration)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
return s_impl.TakeLast(source, duration);
}
///
/// Returns elements within the specified duration from the end of the observable source sequence, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the end of the sequence.
/// Scheduler to run the timer on.
/// An observable sequence with the elements taken during the specified duration from the end of the source sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a buffer with a length enough to store elements for any window during the lifetime of
/// the source sequence. Upon completion of the source sequence, this buffer is drained on the result sequence. This causes the result elements
/// to be delayed with .
///
public static IObservable TakeLast(this IObservable source, TimeSpan duration, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.TakeLast(source, duration, scheduler);
}
///
/// Returns elements within the specified duration from the end of the observable source sequence, using the specified schedulers to run timers and to drain the collected elements.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the end of the sequence.
/// Scheduler to run the timer on.
/// Scheduler to drain the collected elements.
/// An observable sequence with the elements taken during the specified duration from the end of the source sequence.
/// or or is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a buffer with a length enough to store elements for any window during the lifetime of
/// the source sequence. Upon completion of the source sequence, this buffer is drained on the result sequence. This causes the result elements
/// to be delayed with .
///
public static IObservable TakeLast(this IObservable source, TimeSpan duration, IScheduler timerScheduler, IScheduler loopScheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
if (timerScheduler == null)
throw new ArgumentNullException("timerScheduler");
if (loopScheduler == null)
throw new ArgumentNullException("loopScheduler");
return s_impl.TakeLast(source, duration, timerScheduler, loopScheduler);
}
#endregion
#region + TakeLastBuffer +
///
/// Returns a list with the elements within the specified duration from the end of the observable source sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the end of the sequence.
/// An observable sequence containing a single list with the elements taken during the specified duration from the end of the source sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a buffer with a length enough to store elements for any window during the lifetime of
/// the source sequence. Upon completion of the source sequence, this buffer is produced on the result sequence.
///
public static IObservable> TakeLastBuffer(this IObservable source, TimeSpan duration)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
return s_impl.TakeLastBuffer(source, duration);
}
///
/// Returns a list with the elements within the specified duration from the end of the observable source sequence, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Duration for taking elements from the end of the sequence.
/// Scheduler to run the timer on.
/// An observable sequence containing a single list with the elements taken during the specified duration from the end of the source sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
/// This operator accumulates a buffer with a length enough to store elements for any window during the lifetime of
/// the source sequence. Upon completion of the source sequence, this buffer is produced on the result sequence.
///
public static IObservable> TakeLastBuffer(this IObservable source, TimeSpan duration, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (duration < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("duration");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.TakeLastBuffer(source, duration, scheduler);
}
#endregion
#region + TakeUntil +
///
/// Takes elements for the specified duration until the specified end time.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Time to stop taking elements from the source sequence. If this value is less than or equal to DateTimeOffset.UtcNow, the result stream will complete immediately.
/// An observable sequence with the elements taken until the specified end time.
/// is null.
public static IObservable TakeUntil(this IObservable source, DateTimeOffset endTime)
{
if (source == null)
throw new ArgumentNullException("source");
return s_impl.TakeUntil(source, endTime);
}
///
/// Takes elements for the specified duration until the specified end time, using the specified scheduler to run timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to take elements from.
/// Time to stop taking elements from the source sequence. If this value is less than or equal to DateTimeOffset.UtcNow, the result stream will complete immediately.
/// Scheduler to run the timer on.
/// An observable sequence with the elements taken until the specified end time.
/// or is null.
public static IObservable TakeUntil(this IObservable source, DateTimeOffset endTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.TakeUntil(source, endTime, scheduler);
}
#endregion
#region + Throttle +
///
/// Ignores elements from an observable sequence which are followed by another element within a specified relative time duration.
///
/// The type of the elements in the source sequence.
/// Source sequence to throttle.
/// Throttling duration for each element.
/// The throttled sequence.
/// is null.
/// is less than TimeSpan.Zero.
///
///
/// This operator throttles the source sequence by holding on to each element for the duration specified in . If another
/// element is produced within this time window, the element is dropped and a new timer is started for the current element, repeating this whole
/// process. For streams that never have gaps larger than or equal to between elements, the resulting stream won't
/// produce any elements. In order to reduce the volume of a stream whilst guaranteeing the periodic production of elements, consider using the
/// Observable.Sample set of operators.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing throttling timers to be scheduled
/// that are due immediately. However, this doesn't guarantee all elements will be retained in the result sequence. This is a side-effect of the
/// asynchrony introduced by the scheduler, where the action to forward the current element may not execute immediately, despite the TimeSpan.Zero
/// due time. In such cases, the next element may arrive before the scheduler gets a chance to run the throttling action.
///
///
public static IObservable Throttle(this IObservable source, TimeSpan dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
return s_impl.Throttle(source, dueTime);
}
///
/// Ignores elements from an observable sequence which are followed by another element within a specified relative time duration, using the specified scheduler to run throttling timers.
///
/// The type of the elements in the source sequence.
/// Source sequence to throttle.
/// Throttling duration for each element.
/// Scheduler to run the throttle timers on.
/// The throttled sequence.
/// or is null.
/// is less than TimeSpan.Zero.
///
///
/// This operator throttles the source sequence by holding on to each element for the duration specified in . If another
/// element is produced within this time window, the element is dropped and a new timer is started for the current element, repeating this whole
/// process. For streams that never have gaps larger than or equal to between elements, the resulting stream won't
/// produce any elements. In order to reduce the volume of a stream whilst guaranteeing the periodic production of elements, consider using the
/// Observable.Sample set of operators.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing throttling timers to be scheduled
/// that are due immediately. However, this doesn't guarantee all elements will be retained in the result sequence. This is a side-effect of the
/// asynchrony introduced by the scheduler, where the action to forward the current element may not execute immediately, despite the TimeSpan.Zero
/// due time. In such cases, the next element may arrive before the scheduler gets a chance to run the throttling action.
///
///
public static IObservable Throttle(this IObservable source, TimeSpan dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Throttle(source, dueTime, scheduler);
}
///
/// Ignores elements from an observable sequence which are followed by another value within a computed throttle duration.
///
/// The type of the elements in the source sequence.
/// The type of the elements in the throttle sequences selected for each element in the source sequence.
/// Source sequence to throttle.
/// Selector function to retrieve a sequence indicating the throttle duration for each given element.
/// The throttled sequence.
/// or is null.
///
/// This operator throttles the source sequence by holding on to each element for the duration denoted by .
/// If another element is produced within this time window, the element is dropped and a new timer is started for the current element, repeating this
/// whole process. For streams where the duration computed by applying the to each element overlaps with
/// the occurrence of the successor element, the resulting stream won't produce any elements. In order to reduce the volume of a stream whilst
/// guaranteeing the periodic production of elements, consider using the Observable.Sample set of operators.
///
public static IObservable Throttle(this IObservable source, Func> throttleDurationSelector)
{
if (source == null)
throw new ArgumentNullException("source");
if (throttleDurationSelector == null)
throw new ArgumentNullException("throttleDurationSelector");
return s_impl.Throttle(source, throttleDurationSelector);
}
#endregion
#region + TimeInterval +
///
/// Records the time interval between consecutive elements in an observable sequence.
///
/// The type of the elements in the source sequence.
/// Source sequence to record time intervals for.
/// An observable sequence with time interval information on elements.
/// is null.
public static IObservable> TimeInterval(this IObservable source)
{
if (source == null)
throw new ArgumentNullException("source");
return s_impl.TimeInterval(source);
}
///
/// Records the time interval between consecutive elements in an observable sequence, using the specified scheduler to compute time intervals.
///
/// The type of the elements in the source sequence.
/// Source sequence to record time intervals for.
/// Scheduler used to compute time intervals.
/// An observable sequence with time interval information on elements.
/// or is null.
public static IObservable> TimeInterval(this IObservable source, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.TimeInterval(source, scheduler);
}
#endregion
#region + Timeout +
#region TimeSpan
///
/// Applies a timeout policy for each element in the observable sequence.
/// If the next element isn't received within the specified timeout duration starting from its predecessor, a TimeoutException is propagated to the observer.
///
/// The type of the elements in the source sequence.
/// Source sequence to perform a timeout for.
/// Maximum duration between values before a timeout occurs.
/// The source sequence with a TimeoutException in case of a timeout.
/// is null.
/// is less than TimeSpan.Zero.
/// (Asynchronous) If no element is produced within from the previous element.
///
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing timeout timers to be scheduled that are due
/// immediately. However, this doesn't guarantee a timeout will occur, even for the first element. This is a side-effect of the asynchrony introduced by the
/// scheduler, where the action to propagate a timeout may not execute immediately, despite the TimeSpan.Zero due time. In such cases, the next element may
/// arrive before the scheduler gets a chance to run the timeout action.
///
///
public static IObservable Timeout(this IObservable source, TimeSpan dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
return s_impl.Timeout(source, dueTime);
}
///
/// Applies a timeout policy for each element in the observable sequence, using the specified scheduler to run timeout timers.
/// If the next element isn't received within the specified timeout duration starting from its predecessor, a TimeoutException is propagated to the observer.
///
/// The type of the elements in the source sequence.
/// Source sequence to perform a timeout for.
/// Maximum duration between values before a timeout occurs.
/// Scheduler to run the timeout timers on.
/// The source sequence with a TimeoutException in case of a timeout.
/// or is null.
/// is less than TimeSpan.Zero.
/// (Asynchronous) If no element is produced within from the previous element.
///
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing timeout timers to be scheduled that are due
/// immediately. However, this doesn't guarantee a timeout will occur, even for the first element. This is a side-effect of the asynchrony introduced by the
/// scheduler, where the action to propagate a timeout may not execute immediately, despite the TimeSpan.Zero due time. In such cases, the next element may
/// arrive before the scheduler gets a chance to run the timeout action.
///
///
public static IObservable Timeout(this IObservable source, TimeSpan dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Timeout(source, dueTime, scheduler);
}
///
/// Applies a timeout policy for each element in the observable sequence.
/// If the next element isn't received within the specified timeout duration starting from its predecessor, the other observable sequence is used to produce future messages from that point on.
///
/// The type of the elements in the source sequence and the other sequence used upon a timeout.
/// Source sequence to perform a timeout for.
/// Maximum duration between values before a timeout occurs.
/// Sequence to return in case of a timeout.
/// The source sequence switching to the other sequence in case of a timeout.
/// or is null.
/// is less than TimeSpan.Zero.
///
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing timeout timers to be scheduled that are due
/// immediately. However, this doesn't guarantee a timeout will occur, even for the first element. This is a side-effect of the asynchrony introduced by the
/// scheduler, where the action to propagate a timeout may not execute immediately, despite the TimeSpan.Zero due time. In such cases, the next element may
/// arrive before the scheduler gets a chance to run the timeout action.
///
///
public static IObservable Timeout(this IObservable source, TimeSpan dueTime, IObservable other)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
if (other == null)
throw new ArgumentNullException("other");
return s_impl.Timeout(source, dueTime, other);
}
///
/// Applies a timeout policy for each element in the observable sequence, using the specified scheduler to run timeout timers.
/// If the next element isn't received within the specified timeout duration starting from its predecessor, the other observable sequence is used to produce future messages from that point on.
///
/// The type of the elements in the source sequence and the other sequence used upon a timeout.
/// Source sequence to perform a timeout for.
/// Maximum duration between values before a timeout occurs.
/// Sequence to return in case of a timeout.
/// Scheduler to run the timeout timers on.
/// The source sequence switching to the other sequence in case of a timeout.
/// or or is null.
/// is less than TimeSpan.Zero.
///
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
///
/// Specifying a TimeSpan.Zero value for is not recommended but supported, causing timeout timers to be scheduled that are due
/// immediately. However, this doesn't guarantee a timeout will occur, even for the first element. This is a side-effect of the asynchrony introduced by the
/// scheduler, where the action to propagate a timeout may not execute immediately, despite the TimeSpan.Zero due time. In such cases, the next element may
/// arrive before the scheduler gets a chance to run the timeout action.
///
///
public static IObservable Timeout(this IObservable source, TimeSpan dueTime, IObservable other, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (dueTime < TimeSpan.Zero)
throw new ArgumentOutOfRangeException("dueTime");
if (other == null)
throw new ArgumentNullException("other");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Timeout(source, dueTime, other, scheduler);
}
#endregion
#region DateTimeOffset
///
/// Applies a timeout policy to the observable sequence based on an absolute time.
/// If the sequence doesn't terminate before the specified absolute due time, a TimeoutException is propagated to the observer.
///
/// The type of the elements in the source sequence.
/// Source sequence to perform a timeout for.
/// Time when a timeout occurs. If this value is less than or equal to DateTimeOffset.UtcNow, the timeout occurs immediately.
/// The source sequence with a TimeoutException in case of a timeout.
/// is null.
/// (Asynchronous) If the sequence hasn't terminated before .
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
public static IObservable Timeout(this IObservable source, DateTimeOffset dueTime)
{
if (source == null)
throw new ArgumentNullException("source");
return s_impl.Timeout(source, dueTime);
}
///
/// Applies a timeout policy to the observable sequence based on an absolute time, using the specified scheduler to run timeout timers.
/// If the sequence doesn't terminate before the specified absolute due time, a TimeoutException is propagated to the observer.
///
/// The type of the elements in the source sequence.
/// Source sequence to perform a timeout for.
/// Time when a timeout occurs. If this value is less than or equal to DateTimeOffset.UtcNow, the timeout occurs immediately.
/// Scheduler to run the timeout timers on.
/// The source sequence with a TimeoutException in case of a timeout.
/// or is null.
/// (Asynchronous) If the sequence hasn't terminated before .
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
public static IObservable Timeout(this IObservable source, DateTimeOffset dueTime, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return s_impl.Timeout(source, dueTime, scheduler);
}
///
/// Applies a timeout policy to the observable sequence based on an absolute time.
/// If the sequence doesn't terminate before the specified absolute due time, the other observable sequence is used to produce future messages from that point on.
///
/// The type of the elements in the source sequence and the other sequence used upon a timeout.
/// Source sequence to perform a timeout for.
/// Time when a timeout occurs. If this value is less than or equal to DateTimeOffset.UtcNow, the timeout occurs immediately.
/// Sequence to return in case of a timeout.
/// The source sequence switching to the other sequence in case of a timeout.
/// or is null.
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
public static IObservable Timeout(this IObservable source, DateTimeOffset dueTime, IObservable other)
{
if (source == null)
throw new ArgumentNullException("source");
if (other == null)
throw new ArgumentNullException("other");
return s_impl.Timeout(source, dueTime, other);
}
///
/// Applies a timeout policy to the observable sequence based on an absolute time, using the specified scheduler to run timeout timers.
/// If the sequence doesn't terminate before the specified absolute due time, the other observable sequence is used to produce future messages from that point on.
///
/// The type of the elements in the source sequence and the other sequence used upon a timeout.
/// Source sequence to perform a timeout for.
/// Time when a timeout occurs. If this value is less than or equal to DateTimeOffset.UtcNow, the timeout occurs immediately.
/// Sequence to return in case of a timeout.
/// Scheduler to run the timeout timers on.
/// The source sequence switching to the other sequence in case of a timeout.
/// or or is null.
///
/// In case you only want to timeout on the first element, consider using the
/// operator applied to the source sequence and a delayed sequence. Alternatively, the general-purpose overload
/// of Timeout, can be used.
///
public static IObservable Timeout(this IObservable source, DateTimeOffset dueTime, IObservable other, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
if (other == null)
throw new ArgumentNullException("other");
return s_impl.Timeout(source, dueTime, other, scheduler);
}
#endregion
#region Duration selector
///
/// Applies a timeout policy to the observable sequence based on a timeout duration computed for each element.
/// If the next element isn't received within the computed duration starting from its predecessor, a TimeoutException is propagated to the observer.
///
/// The type of the elements in the source sequence.
/// The type of the elements in the timeout sequences used to indicate the timeout duration for each element in the source sequence.
/// Source sequence to perform a timeout for.
/// Selector to retrieve an observable sequence that represents the timeout between the current element and the next element.
/// The source sequence with a TimeoutException in case of a timeout.
/// or is null.
public static IObservable Timeout(this IObservable source, Func> timeoutDurationSelector)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeoutDurationSelector == null)
throw new ArgumentNullException("timeoutDurationSelector");
return s_impl.Timeout(source, timeoutDurationSelector);
}
///
/// Applies a timeout policy to the observable sequence based on a timeout duration computed for each element.
/// If the next element isn't received within the computed duration starting from its predecessor, the other observable sequence is used to produce future messages from that point on.
///
/// The type of the elements in the source sequence and the other sequence used upon a timeout.
/// The type of the elements in the timeout sequences used to indicate the timeout duration for each element in the source sequence.
/// Source sequence to perform a timeout for.
/// Selector to retrieve an observable sequence that represents the timeout between the current element and the next element.
/// Sequence to return in case of a timeout.
/// The source sequence switching to the other sequence in case of a timeout.
/// or or is null.
public static IObservable Timeout(this IObservable source, Func> timeoutDurationSelector, IObservable other)
{
if (source == null)
throw new ArgumentNullException("source");
if (timeoutDurationSelector == null)
throw new ArgumentNullException("timeoutDurationSelector");
if (other == null)
throw new ArgumentNullException("other");
return s_impl.Timeout(source, timeoutDurationSelector, other);
}
///
/// Applies a timeout policy to the observable sequence based on an initial timeout duration for the first element, and a timeout duration computed for each subsequent element.
/// If the next element isn't received within the computed duration starting from its predecessor, a TimeoutException is propagated to the observer.
///
/// The type of the elements in the source sequence.
/// The type of the elements in the timeout sequences used to indicate the timeout duration for each element in the source sequence.
/// Source sequence to perform a timeout for.
/// Observable sequence that represents the timeout for the first element.
/// Selector to retrieve an observable sequence that represents the timeout between the current element and the next element.
/// The source sequence with a TimeoutException in case of a timeout.
/// or or is null.
public static IObservable Timeout(this IObservable source, IObservable firstTimeout, Func> timeoutDurationSelector)
{
if (source == null)
throw new ArgumentNullException("source");
if (firstTimeout == null)
throw new ArgumentNullException("firstTimeout");
if (timeoutDurationSelector == null)
throw new ArgumentNullException("timeoutDurationSelector");
return s_impl.Timeout(source, firstTimeout, timeoutDurationSelector);
}
///
/// Applies a timeout policy to the observable sequence based on an initial timeout duration for the first element, and a timeout duration computed for each subsequent element.
/// If the next element isn't received within the computed duration starting from its predecessor, the other observable sequence is used to produce future messages from that point on.
///
/// The type of the elements in the source sequence and the other sequence used upon a timeout.
/// The type of the elements in the timeout sequences used to indicate the timeout duration for each element in the source sequence.
/// Source sequence to perform a timeout for.
/// Observable sequence that represents the timeout for the first element.
/// Selector to retrieve an observable sequence that represents the timeout between the current element and the next element.
/// Sequence to return in case of a timeout.
/// The source sequence switching to the other sequence in case of a timeout.
/// or or or is null.
public static IObservable Timeout(this IObservable source, IObservable firstTimeout, Func> timeoutDurationSelector, IObservable other)
{
if (source == null)
throw new ArgumentNullException("source");
if (firstTimeout == null)
throw new ArgumentNullException("firstTimeout");
if (timeoutDurationSelector == null)
throw new ArgumentNullException("timeoutDurationSelector");
if (other == null)
throw new ArgumentNullException("other");
return s_impl.Timeout(source, firstTimeout, timeoutDurationSelector, other);
}
#endregion
#endregion
#region + Timer +
///
/// Returns an observable sequence that produces a single value after the specified relative due time has elapsed.
///
/// Relative time at which to produce the value. If this value is less than or equal to TimeSpan.Zero, the timer will fire as soon as possible.
/// An observable sequence that produces a value after the due time has elapsed.
public static IObservable Timer(TimeSpan dueTime)
{
return s_impl.Timer(dueTime);
}
///
/// Returns an observable sequence that produces a single value at the specified absolute due time.
///
/// Absolute time at which to produce the value. If this value is less than or equal to DateTimeOffset.UtcNow, the timer will fire as soon as possible.
/// An observable sequence that produces a value at due time.
public static IObservable Timer(DateTimeOffset dueTime)
{
return s_impl.Timer(dueTime);
}
///