using System.Reactive.Concurrency; using System.Reactive.Disposables; using System.Runtime.CompilerServices; namespace System.Reactive.Linq.ObservableImpl { [System.Runtime.CompilerServices.NullableContext(1)] [System.Runtime.CompilerServices.Nullable(new byte[] { 0, 1, 1, 1 })] internal sealed class Throttle<[System.Runtime.CompilerServices.Nullable(2)] TSource> : Producer<TSource, Throttle<TSource>._> { [System.Runtime.CompilerServices.Nullable(new byte[] { 0, 1 })] internal sealed class _ : IdentitySink<TSource> { private readonly object _gate = new object(); private readonly TimeSpan _dueTime; private readonly IScheduler _scheduler; [System.Runtime.CompilerServices.Nullable(2)] private TSource _value; private bool _hasValue; private SerialDisposableValue _serialCancelable; private ulong _id; public _(Throttle<TSource> parent, IObserver<TSource> observer) : base(observer) { _dueTime = parent._dueTime; _scheduler = parent._scheduler; } protected override void Dispose(bool disposing) { if (disposing) _serialCancelable.Dispose(); base.Dispose(disposing); } public override void OnNext(TSource value) { ulong id = default(ulong); lock (_gate) { _hasValue = true; _value = value; _id++; id = _id; } _serialCancelable.Disposable = null; _serialCancelable.Disposable = Scheduler.ScheduleAction<(_, ulong)>(_scheduler, (this, id), _dueTime, (Action<(_, ulong)>)delegate((_ this, ulong currentid) tuple) { tuple.this.Propagate(tuple.currentid); }); } private void Propagate(ulong currentid) { lock (_gate) { if (_hasValue && _id == currentid) { ForwardOnNext(_value); _hasValue = false; } } } public override void OnError(Exception error) { _serialCancelable.Dispose(); lock (_gate) { ForwardOnError(error); _hasValue = false; _id++; } } public override void OnCompleted() { _serialCancelable.Dispose(); lock (_gate) { if (_hasValue) ForwardOnNext(_value); ForwardOnCompleted(); _hasValue = false; _id++; } } } private readonly IObservable<TSource> _source; private readonly TimeSpan _dueTime; private readonly IScheduler _scheduler; public Throttle(IObservable<TSource> source, TimeSpan dueTime, IScheduler scheduler) { _source = source; _dueTime = dueTime; _scheduler = scheduler; } [return: System.Runtime.CompilerServices.Nullable(new byte[] { 1, 0 })] protected override _ CreateSink(IObserver<TSource> observer) { return new _(this, observer); } protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] { 1, 0 })] _ sink) { sink.Run(_source); } } }