using System;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Runtime.CompilerServices;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Throttle<TSource> : Producer<TSource, Throttle<TSource>._>
	{
		public Throttle(IObservable<TSource> source, TimeSpan dueTime, IScheduler scheduler)
		{
			this._source = source;
			this._dueTime = dueTime;
			this._scheduler = scheduler;
		}

		protected override Throttle<TSource>._ CreateSink(IObserver<TSource> observer)
		{
			return new Throttle<TSource>._(this, observer);
		}

		protected override void Run(Throttle<TSource>._ sink)
		{
			sink.Run(this._source);
		}

		private readonly IObservable<TSource> _source;

		private readonly TimeSpan _dueTime;

		private readonly IScheduler _scheduler;

		internal sealed class _ : IdentitySink<TSource>
		{
			public _(Throttle<TSource> parent, IObserver<TSource> observer)
				: base(observer)
			{
				this._dueTime = parent._dueTime;
				this._scheduler = parent._scheduler;
			}

			protected override void Dispose(bool disposing)
			{
				if (disposing)
				{
					Disposable.TryDispose(ref this._serialCancelable);
				}
				base.Dispose(disposing);
			}

			public override void OnNext(TSource value)
			{
				ulong num = 0UL;
				object gate = this._gate;
				lock (gate)
				{
					this._hasValue = true;
					this._value = value;
					this._id += 1UL;
					num = this._id;
				}
				Disposable.TrySetSerial(ref this._serialCancelable, null);
				Disposable.TrySetSerial(ref this._serialCancelable, this._scheduler.ScheduleAction(new ValueTuple<Throttle<TSource>._, ulong>(this, num), this._dueTime, delegate([TupleElementNames(new string[] { "this", "currentid" })] ValueTuple<Throttle<TSource>._, ulong> tuple)
				{
					tuple.Item1.Propagate(tuple.Item2);
				}));
			}

			private void Propagate(ulong currentid)
			{
				object gate = this._gate;
				lock (gate)
				{
					if (this._hasValue && this._id == currentid)
					{
						base.ForwardOnNext(this._value);
					}
					this._hasValue = false;
				}
			}

			public override void OnError(Exception error)
			{
				Disposable.TryDispose(ref this._serialCancelable);
				object gate = this._gate;
				lock (gate)
				{
					base.ForwardOnError(error);
					this._hasValue = false;
					this._id += 1UL;
				}
			}

			public override void OnCompleted()
			{
				Disposable.TryDispose(ref this._serialCancelable);
				object gate = this._gate;
				lock (gate)
				{
					if (this._hasValue)
					{
						base.ForwardOnNext(this._value);
					}
					base.ForwardOnCompleted();
					this._hasValue = false;
					this._id += 1UL;
				}
			}

			private readonly TimeSpan _dueTime;

			private readonly IScheduler _scheduler;

			private readonly object _gate = new object();

			private TSource _value;

			private bool _hasValue;

			private IDisposable _serialCancelable;

			private ulong _id;
		}
	}
}
