using System;
using System.Collections.Generic;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Threading;

namespace System.Reactive.Linq.ObservableImpl
{
	internal static class Delay<TSource>
	{
		internal abstract class Base<TParent> : Producer<TSource, Delay<TSource>.Base<TParent>._> where TParent : Delay<TSource>.Base<TParent>
		{
			protected Base(IObservable<TSource> source, IScheduler scheduler)
			{
				this._source = source;
				this._scheduler = scheduler;
			}

			protected readonly IObservable<TSource> _source;

			protected readonly IScheduler _scheduler;

			internal abstract class _ : IdentitySink<TSource>
			{
				protected _(TParent parent, IObserver<TSource> observer)
					: base(observer)
				{
					this._scheduler = parent._scheduler;
				}

				public void Run(TParent parent)
				{
					this._watch = this._scheduler.StartStopwatch();
					this.RunCore(parent);
					base.Run(parent._source);
				}

				protected abstract void RunCore(TParent parent);

				protected IStopwatch _watch;

				protected IScheduler _scheduler;
			}

			internal abstract class S : Delay<TSource>.Base<TParent>._
			{
				protected S(TParent parent, IObserver<TSource> observer)
					: base(parent, observer)
				{
				}

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

				public override void OnNext(TSource value)
				{
					bool flag = false;
					object gate = this._gate;
					lock (gate)
					{
						TimeSpan timeSpan = this._watch.Elapsed.Add(this._delay);
						this._queue.Enqueue(new TimeInterval<TSource>(value, timeSpan));
						flag = this._ready && !this._active;
						this._active = true;
					}
					if (flag)
					{
						Disposable.TrySetSerial(ref this._cancelable, this._scheduler.Schedule(this, this._delay, delegate(Delay<TSource>.Base<TParent>.S @this, Action<Delay<TSource>.Base<TParent>.S, TimeSpan> a)
						{
							@this.DrainQueue(a);
						}));
					}
				}

				public override void OnError(Exception error)
				{
					base.DisposeUpstream();
					bool flag = false;
					object gate = this._gate;
					lock (gate)
					{
						this._queue.Clear();
						this._exception = error;
						this._hasFailed = true;
						flag = !this._running;
					}
					if (flag)
					{
						base.ForwardOnError(error);
					}
				}

				public override void OnCompleted()
				{
					base.DisposeUpstream();
					bool flag = false;
					object gate = this._gate;
					lock (gate)
					{
						TimeSpan timeSpan = this._watch.Elapsed.Add(this._delay);
						this._completeAt = timeSpan;
						this._hasCompleted = true;
						flag = this._ready && !this._active;
						this._active = true;
					}
					if (flag)
					{
						Disposable.TrySetSerial(ref this._cancelable, this._scheduler.Schedule(this, this._delay, delegate(Delay<TSource>.Base<TParent>.S @this, Action<Delay<TSource>.Base<TParent>.S, TimeSpan> a)
						{
							@this.DrainQueue(a);
						}));
					}
				}

				protected void DrainQueue(Action<Delay<TSource>.Base<TParent>.S, TimeSpan> recurse)
				{
					object obj = this._gate;
					lock (obj)
					{
						if (this._hasFailed)
						{
							return;
						}
						this._running = true;
					}
					bool flag2 = false;
					bool flag3;
					Exception ex;
					bool flag5;
					bool flag6;
					TimeSpan timeSpan;
					for (;;)
					{
						flag3 = false;
						ex = null;
						bool flag4 = false;
						TSource tsource = default(TSource);
						flag5 = false;
						flag6 = false;
						timeSpan = default(TimeSpan);
						obj = this._gate;
						lock (obj)
						{
							if (this._hasFailed)
							{
								ex = this._exception;
								flag3 = true;
								this._running = false;
								goto IL_0163;
							}
							TimeSpan elapsed = this._watch.Elapsed;
							if (this._queue.Count > 0)
							{
								TimeSpan interval = this._queue.Peek().Interval;
								if (interval.CompareTo(elapsed) <= 0 && !flag2)
								{
									tsource = this._queue.Dequeue().Value;
									flag4 = true;
									goto IL_0163;
								}
								flag6 = true;
								timeSpan = Scheduler.Normalize(interval.Subtract(elapsed));
								this._running = false;
								goto IL_0163;
							}
							else
							{
								if (!this._hasCompleted)
								{
									this._running = false;
									this._active = false;
									goto IL_0163;
								}
								if (this._completeAt.CompareTo(elapsed) <= 0 && !flag2)
								{
									flag5 = true;
									goto IL_0163;
								}
								flag6 = true;
								timeSpan = Scheduler.Normalize(this._completeAt.Subtract(elapsed));
								this._running = false;
								goto IL_0163;
							}
						}
						IL_0154:
						base.ForwardOnNext(tsource);
						flag2 = true;
						continue;
						IL_0163:
						if (!flag4)
						{
							break;
						}
						goto IL_0154;
					}
					if (flag5)
					{
						base.ForwardOnCompleted();
						return;
					}
					if (flag3)
					{
						base.ForwardOnError(ex);
						return;
					}
					if (flag6)
					{
						recurse(this, timeSpan);
					}
				}

				protected readonly object _gate = new object();

				protected IDisposable _cancelable;

				protected TimeSpan _delay;

				protected bool _ready;

				protected bool _active;

				protected bool _running;

				protected Queue<TimeInterval<TSource>> _queue = new Queue<TimeInterval<TSource>>();

				private bool _hasCompleted;

				private TimeSpan _completeAt;

				private bool _hasFailed;

				private Exception _exception;
			}

			protected abstract class L : Delay<TSource>.Base<TParent>._
			{
				protected L(TParent parent, IObserver<TSource> observer)
					: base(parent, observer)
				{
				}

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

				protected void ScheduleDrain()
				{
					this._stop = new CancellationTokenSource();
					Disposable.TrySetSerial(ref this._cancelable, new CancellationDisposable(this._stop));
					this._scheduler.AsLongRunning().ScheduleLongRunning(new Action<ICancelable>(this.DrainQueue));
				}

				public override void OnNext(TSource value)
				{
					object gate = this._gate;
					lock (gate)
					{
						TimeSpan timeSpan = this._watch.Elapsed.Add(this._delay);
						this._queue.Enqueue(new TimeInterval<TSource>(value, timeSpan));
						this._evt.Release();
					}
				}

				public override void OnError(Exception error)
				{
					base.DisposeUpstream();
					object gate = this._gate;
					lock (gate)
					{
						this._queue.Clear();
						this._exception = error;
						this._hasFailed = true;
						this._evt.Release();
					}
				}

				public override void OnCompleted()
				{
					base.DisposeUpstream();
					object gate = this._gate;
					lock (gate)
					{
						TimeSpan timeSpan = this._watch.Elapsed.Add(this._delay);
						this._completeAt = timeSpan;
						this._hasCompleted = true;
						this._evt.Release();
					}
				}

				private void DrainQueue(ICancelable cancel)
				{
					bool flag;
					bool flag3;
					for (;;)
					{
						try
						{
							this._evt.Wait(this._stop.Token);
						}
						catch (OperationCanceledException)
						{
							break;
						}
						flag = false;
						Exception ex = null;
						bool flag2 = false;
						TSource tsource = default(TSource);
						flag3 = false;
						bool flag4 = false;
						TimeSpan timeSpan = default(TimeSpan);
						object gate = this._gate;
						lock (gate)
						{
							if (this._hasFailed)
							{
								ex = this._exception;
								flag = true;
								goto IL_0160;
							}
							TimeSpan elapsed = this._watch.Elapsed;
							if (this._queue.Count > 0)
							{
								TimeInterval<TSource> timeInterval = this._queue.Dequeue();
								flag2 = true;
								tsource = timeInterval.Value;
								TimeSpan interval = timeInterval.Interval;
								if (interval.CompareTo(elapsed) > 0)
								{
									flag4 = true;
									timeSpan = Scheduler.Normalize(interval.Subtract(elapsed));
								}
							}
							else if (this._hasCompleted)
							{
								flag3 = true;
								if (this._completeAt.CompareTo(elapsed) > 0)
								{
									flag4 = true;
									timeSpan = Scheduler.Normalize(this._completeAt.Subtract(elapsed));
								}
							}
							goto IL_0160;
						}
						goto IL_0104;
						IL_0151:
						if (flag2)
						{
							base.ForwardOnNext(tsource);
							continue;
						}
						goto IL_0168;
						IL_0104:
						ManualResetEventSlim manualResetEventSlim = new ManualResetEventSlim();
						this._scheduler.ScheduleAction(manualResetEventSlim, timeSpan, delegate(ManualResetEventSlim slimTimer)
						{
							slimTimer.Set();
						});
						try
						{
							manualResetEventSlim.Wait(this._stop.Token);
						}
						catch (OperationCanceledException)
						{
							break;
						}
						goto IL_0151;
						IL_0160:
						if (flag4)
						{
							goto IL_0104;
						}
						goto IL_0151;
					}
					return;
					IL_0168:
					if (flag3)
					{
						base.ForwardOnCompleted();
						return;
					}
					if (flag)
					{
						Exception ex;
						base.ForwardOnError(ex);
					}
				}

				protected readonly object _gate = new object();

				protected IDisposable _cancelable;

				private readonly SemaphoreSlim _evt = new SemaphoreSlim(0);

				protected TimeSpan _delay;

				protected Queue<TimeInterval<TSource>> _queue = new Queue<TimeInterval<TSource>>();

				private CancellationTokenSource _stop;

				private bool _hasCompleted;

				private TimeSpan _completeAt;

				private bool _hasFailed;

				private Exception _exception;
			}
		}

		internal sealed class Absolute : Delay<TSource>.Base<Delay<TSource>.Absolute>
		{
			public Absolute(IObservable<TSource> source, DateTimeOffset dueTime, IScheduler scheduler)
				: base(source, scheduler)
			{
				this._dueTime = dueTime;
			}

			protected override Delay<TSource>.Base<Delay<TSource>.Absolute>._ CreateSink(IObserver<TSource> observer)
			{
				if (this._scheduler.AsLongRunning() == null)
				{
					return new Delay<TSource>.Absolute.S(this, observer);
				}
				return new Delay<TSource>.Absolute.L(this, observer);
			}

			protected override void Run(Delay<TSource>.Base<Delay<TSource>.Absolute>._ sink)
			{
				sink.Run(this);
			}

			private readonly DateTimeOffset _dueTime;

			private new sealed class S : Delay<TSource>.Base<Delay<TSource>.Absolute>.S
			{
				public S(Delay<TSource>.Absolute parent, IObserver<TSource> observer)
					: base(parent, observer)
				{
				}

				protected override void RunCore(Delay<TSource>.Absolute parent)
				{
					this._ready = false;
					Disposable.TrySetSingle(ref this._cancelable, parent._scheduler.ScheduleAction(this, parent._dueTime, delegate(Delay<TSource>.Absolute.S @this)
					{
						@this.Start();
					}));
				}

				private void Start()
				{
					TimeSpan timeSpan = default(TimeSpan);
					bool flag = false;
					object gate = this._gate;
					lock (gate)
					{
						this._delay = this._watch.Elapsed;
						Queue<TimeInterval<TSource>> queue = this._queue;
						this._queue = new Queue<TimeInterval<TSource>>();
						if (queue.Count > 0)
						{
							timeSpan = queue.Peek().Interval;
							while (queue.Count > 0)
							{
								TimeInterval<TSource> timeInterval = queue.Dequeue();
								this._queue.Enqueue(new TimeInterval<TSource>(timeInterval.Value, timeInterval.Interval.Add(this._delay)));
							}
							flag = true;
							this._active = true;
						}
						this._ready = true;
					}
					if (flag)
					{
						Disposable.TrySetSerial(ref this._cancelable, this._scheduler.Schedule(this, timeSpan, delegate(Delay<TSource>.Base<Delay<TSource>.Absolute>.S @this, Action<Delay<TSource>.Base<Delay<TSource>.Absolute>.S, TimeSpan> a)
						{
							base.DrainQueue(a);
						}));
					}
				}
			}

			private new sealed class L : Delay<TSource>.Base<Delay<TSource>.Absolute>.L
			{
				public L(Delay<TSource>.Absolute parent, IObserver<TSource> observer)
					: base(parent, observer)
				{
				}

				protected override void RunCore(Delay<TSource>.Absolute parent)
				{
					Disposable.TrySetSingle(ref this._cancelable, parent._scheduler.ScheduleAction(this, parent._dueTime, delegate(Delay<TSource>.Absolute.L @this)
					{
						@this.Start();
					}));
				}

				private void Start()
				{
					object gate = this._gate;
					lock (gate)
					{
						this._delay = this._watch.Elapsed;
						Queue<TimeInterval<TSource>> queue = this._queue;
						this._queue = new Queue<TimeInterval<TSource>>();
						while (queue.Count > 0)
						{
							TimeInterval<TSource> timeInterval = queue.Dequeue();
							this._queue.Enqueue(new TimeInterval<TSource>(timeInterval.Value, timeInterval.Interval.Add(this._delay)));
						}
					}
					base.ScheduleDrain();
				}
			}
		}

		internal sealed class Relative : Delay<TSource>.Base<Delay<TSource>.Relative>
		{
			public Relative(IObservable<TSource> source, TimeSpan dueTime, IScheduler scheduler)
				: base(source, scheduler)
			{
				this._dueTime = dueTime;
			}

			protected override Delay<TSource>.Base<Delay<TSource>.Relative>._ CreateSink(IObserver<TSource> observer)
			{
				if (this._scheduler.AsLongRunning() == null)
				{
					return new Delay<TSource>.Relative.S(this, observer);
				}
				return new Delay<TSource>.Relative.L(this, observer);
			}

			protected override void Run(Delay<TSource>.Base<Delay<TSource>.Relative>._ sink)
			{
				sink.Run(this);
			}

			private readonly TimeSpan _dueTime;

			private new sealed class S : Delay<TSource>.Base<Delay<TSource>.Relative>.S
			{
				public S(Delay<TSource>.Relative parent, IObserver<TSource> observer)
					: base(parent, observer)
				{
				}

				protected override void RunCore(Delay<TSource>.Relative parent)
				{
					this._ready = true;
					this._delay = Scheduler.Normalize(parent._dueTime);
				}
			}

			private new sealed class L : Delay<TSource>.Base<Delay<TSource>.Relative>.L
			{
				public L(Delay<TSource>.Relative parent, IObserver<TSource> observer)
					: base(parent, observer)
				{
				}

				protected override void RunCore(Delay<TSource>.Relative parent)
				{
					this._delay = Scheduler.Normalize(parent._dueTime);
					base.ScheduleDrain();
				}
			}
		}
	}
}
