using System;
using System.Reactive.Disposables;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Throttle<TSource, T> : Producer<TSource, Throttle<TSource, T>._>
	{
		public Throttle(IObservable<TSource> source, Func<TSource, IObservable<T>> throttleSelector)
		{
			this._source = source;
			this._throttleSelector = throttleSelector;
		}

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

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

		private readonly IObservable<TSource> _source;

		private readonly Func<TSource, IObservable<T>> _throttleSelector;

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

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

			public override void OnNext(TSource value)
			{
				IObservable<T> observable = null;
				object obj;
				try
				{
					observable = this._throttleSelector(value);
				}
				catch (Exception ex)
				{
					obj = this._gate;
					lock (obj)
					{
						base.ForwardOnError(ex);
					}
					return;
				}
				obj = this._gate;
				ulong id;
				lock (obj)
				{
					this._hasValue = true;
					this._value = value;
					this._id += 1UL;
					id = this._id;
				}
				Disposable.TrySetSerial(ref this._serialCancelable, null);
				Throttle<TSource, T>._.ThrottleObserver throttleObserver = new Throttle<TSource, T>._.ThrottleObserver(this, value, id);
				throttleObserver.SetResource(observable.SubscribeSafe(throttleObserver));
				Disposable.TrySetSerial(ref this._serialCancelable, throttleObserver);
			}

			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 Func<TSource, IObservable<T>> _throttleSelector;

			private readonly object _gate = new object();

			private TSource _value;

			private bool _hasValue;

			private IDisposable _serialCancelable;

			private ulong _id;

			private sealed class ThrottleObserver : SafeObserver<T>
			{
				public ThrottleObserver(Throttle<TSource, T>._ parent, TSource value, ulong currentid)
				{
					this._parent = parent;
					this._value = value;
					this._currentid = currentid;
				}

				public override void OnNext(T value)
				{
					object gate = this._parent._gate;
					lock (gate)
					{
						if (this._parent._hasValue && this._parent._id == this._currentid)
						{
							this._parent.ForwardOnNext(this._value);
						}
						this._parent._hasValue = false;
						base.Dispose();
					}
				}

				public override void OnError(Exception error)
				{
					object gate = this._parent._gate;
					lock (gate)
					{
						this._parent.ForwardOnError(error);
					}
				}

				public override void OnCompleted()
				{
					object gate = this._parent._gate;
					lock (gate)
					{
						if (this._parent._hasValue && this._parent._id == this._currentid)
						{
							this._parent.ForwardOnNext(this._value);
						}
						this._parent._hasValue = false;
						base.Dispose();
					}
				}

				private readonly Throttle<TSource, T>._ _parent;

				private readonly TSource _value;

				private readonly ulong _currentid;
			}
		}
	}
}
