Throttle<TSource, TThrottle>
using System.Reactive.Disposables;
namespace System.Reactive.Linq.ObservableImpl
{
internal sealed class Throttle<TSource, TThrottle> : Producer<TSource, Throttle<TSource, TThrottle>._>
{
internal sealed class _ : Sink<TSource>, IObserver<TSource>
{
private sealed class ThrottleObserver : IObserver<TThrottle>
{
private readonly _ _parent;
private readonly TSource _value;
private readonly ulong _currentid;
private readonly IDisposable _self;
public ThrottleObserver(_ parent, TSource value, ulong currentid, IDisposable self)
{
_parent = parent;
_value = value;
_currentid = currentid;
_self = self;
}
public void OnNext(TThrottle value)
{
lock (_parent._gate) {
if (_parent._hasValue && _parent._id == _currentid)
_parent._observer.OnNext(_value);
_parent._hasValue = false;
_self.Dispose();
}
}
public void OnError(Exception error)
{
lock (_parent._gate) {
_parent._observer.OnError(error);
_parent.Dispose();
}
}
public void OnCompleted()
{
lock (_parent._gate) {
if (_parent._hasValue && _parent._id == _currentid)
_parent._observer.OnNext(_value);
_parent._hasValue = false;
_self.Dispose();
}
}
}
private readonly Func<TSource, IObservable<TThrottle>> _throttleSelector;
private object _gate;
private TSource _value;
private bool _hasValue;
private SerialDisposable _cancelable;
private ulong _id;
public _(Throttle<TSource, TThrottle> parent, IObserver<TSource> observer, IDisposable cancel)
: base(observer, cancel)
{
_throttleSelector = parent._throttleSelector;
}
public IDisposable Run(Throttle<TSource, TThrottle> parent)
{
_gate = new object();
_value = default(TSource);
_hasValue = false;
_cancelable = new SerialDisposable();
_id = 0;
return StableCompositeDisposable.Create(ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this), _cancelable);
}
public void OnNext(TSource value)
{
IObservable<TThrottle> observable = null;
try {
observable = _throttleSelector(value);
} catch (Exception error) {
lock (_gate) {
_observer.OnError(error);
base.Dispose();
}
return;
}
ulong id = default(ulong);
lock (_gate) {
_hasValue = true;
_value = value;
_id++;
id = _id;
}
SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable();
_cancelable.Disposable = singleAssignmentDisposable;
singleAssignmentDisposable.Disposable = ObservableExtensions.SubscribeSafe<TThrottle>(observable, (IObserver<TThrottle>)new ThrottleObserver(this, value, id, singleAssignmentDisposable));
}
public void OnError(Exception error)
{
_cancelable.Dispose();
lock (_gate) {
_observer.OnError(error);
base.Dispose();
_hasValue = false;
_id++;
}
}
public void OnCompleted()
{
_cancelable.Dispose();
lock (_gate) {
if (_hasValue)
_observer.OnNext(_value);
_observer.OnCompleted();
base.Dispose();
_hasValue = false;
_id++;
}
}
}
private readonly IObservable<TSource> _source;
private readonly Func<TSource, IObservable<TThrottle>> _throttleSelector;
public Throttle(IObservable<TSource> source, Func<TSource, IObservable<TThrottle>> throttleSelector)
{
_source = source;
_throttleSelector = throttleSelector;
}
protected override _ CreateSink(IObserver<TSource> observer, IDisposable cancel)
{
return new _(this, observer, cancel);
}
protected override IDisposable Run(_ sink)
{
return sink.Run(this);
}
}
}