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

namespace System.Reactive
{
	internal class ScheduledObserver<T> : ObserverBase<T>, Interface3<T>, IObserver<T>, IDisposable
	{
		public ScheduledObserver(IScheduler scheduler, IObserver<T> observer)
		{
			this._scheduler = scheduler;
			this._observer = observer;
			this._longRunning = this._scheduler.AsLongRunning();
			if (this._longRunning != null)
			{
				this._dispatcherEvent = new SemaphoreSlim(0);
				this._dispatcherEventRelease = new ScheduledObserver<T>.SemaphoreSlimRelease(this._dispatcherEvent);
			}
		}

		private void EnsureDispatcher()
		{
			if (this._dispatcherJob == null)
			{
				object dispatcherInitGate = this._dispatcherInitGate;
				lock (dispatcherInitGate)
				{
					if (this._dispatcherJob == null)
					{
						this._dispatcherJob = this._longRunning.ScheduleLongRunning(new Action<ICancelable>(this.Dispatch));
						Disposable.TrySetSerial(ref this._disposable, StableCompositeDisposable.Create(this._dispatcherJob, this._dispatcherEventRelease));
					}
				}
			}
		}

		private void Dispatch(ICancelable cancel)
		{
			do
			{
				this._dispatcherEvent.Wait();
				if (cancel.IsDisposed)
				{
					return;
				}
				T t = default(T);
				while (this._queue.TryDequeue(out t))
				{
					try
					{
						this._observer.OnNext(t);
						goto IL_0063;
					}
					catch
					{
						T t2;
						while (this._queue.TryDequeue(out t2))
						{
						}
						throw;
					}
					continue;
					IL_0063:
					this._dispatcherEvent.Wait();
					if (cancel.IsDisposed)
					{
						return;
					}
				}
				if (this._failed)
				{
					goto IL_008B;
				}
			}
			while (!this._completed);
			this._observer.OnCompleted();
			base.Dispose();
			return;
			IL_008B:
			this._observer.OnError(this._error);
			base.Dispose();
		}

		public void EnsureActive()
		{
			this.EnsureActive(1);
		}

		public void EnsureActive(int n)
		{
			if (this._longRunning != null)
			{
				if (n > 0)
				{
					this._dispatcherEvent.Release(n);
				}
				this.EnsureDispatcher();
				return;
			}
			this.EnsureActiveSlow();
		}

		private void EnsureActiveSlow()
		{
			bool flag = false;
			for (;;)
			{
				int num = Interlocked.CompareExchange(ref this._state, 1, 0);
				if (num == 0)
				{
					goto Block_5;
				}
				if (num == 9)
				{
					break;
				}
				if (num == 2)
				{
					goto IL_003D;
				}
				if (num == 1 && Interlocked.CompareExchange(ref this._state, 2, 1) == 1)
				{
					goto Block_4;
				}
			}
			return;
			Block_4:
			goto IL_003D;
			Block_5:
			flag = true;
			IL_003D:
			if (flag)
			{
				Disposable.TrySetSerial(ref this._disposable, this._scheduler.Schedule(null, new Action<object, Action<object>>(this.Run)));
			}
		}

		private void Run(object state, Action<object> recurse)
		{
			T t = default(T);
			while (!this._queue.TryDequeue(out t))
			{
				if (this._failed)
				{
					if (this._queue.IsEmpty)
					{
						Interlocked.Exchange(ref this._state, 0);
						this._observer.OnError(this._error);
						base.Dispose();
						return;
					}
				}
				else
				{
					if (!this._completed)
					{
						int num = Interlocked.CompareExchange(ref this._state, 0, 1);
						if (num != 1)
						{
							if (num != 9)
							{
								this._state = 1;
								continue;
							}
						}
						return;
					}
					if (this._queue.IsEmpty)
					{
						Interlocked.Exchange(ref this._state, 0);
						this._observer.OnCompleted();
						base.Dispose();
						return;
					}
				}
			}
			Interlocked.Exchange(ref this._state, 1);
			try
			{
				this._observer.OnNext(t);
			}
			catch
			{
				Interlocked.Exchange(ref this._state, 9);
				T t2;
				while (this._queue.TryDequeue(out t2))
				{
				}
				throw;
			}
			recurse(state);
		}

		protected override void OnNextCore(T value)
		{
			this._queue.Enqueue(value);
		}

		protected override void OnErrorCore(Exception exception)
		{
			this._error = exception;
			this._failed = true;
		}

		protected override void OnCompletedCore()
		{
			this._completed = true;
		}

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

		private int _state;

		private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();

		private bool _failed;

		private Exception _error;

		private bool _completed;

		private readonly IObserver<T> _observer;

		private readonly IScheduler _scheduler;

		private readonly ISchedulerLongRunning _longRunning;

		private IDisposable _disposable;

		private readonly object _dispatcherInitGate = new object();

		private readonly SemaphoreSlim _dispatcherEvent;

		private readonly IDisposable _dispatcherEventRelease;

		private IDisposable _dispatcherJob;

		private sealed class SemaphoreSlimRelease : IDisposable
		{
			public SemaphoreSlimRelease(SemaphoreSlim dispatcherEvent)
			{
				Volatile.Write<SemaphoreSlim>(ref this._dispatcherEvent, dispatcherEvent);
			}

			public void Dispose()
			{
				SemaphoreSlim semaphoreSlim = Interlocked.Exchange<SemaphoreSlim>(ref this._dispatcherEvent, null);
				if (semaphoreSlim == null)
				{
					return;
				}
				semaphoreSlim.Release();
			}

			private SemaphoreSlim _dispatcherEvent;
		}
	}
}
