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

namespace System.Reactive
{
	internal sealed class ObserveOnObserverLongRunning<TSource> : IdentitySink<TSource>
	{
		public ObserveOnObserverLongRunning(ISchedulerLongRunning scheduler, IObserver<TSource> observer)
			: base(observer)
		{
			this._scheduler = scheduler;
			this._queue = new ConcurrentQueue<TSource>();
			this._suspendGuard = new object();
		}

		public override void OnCompleted()
		{
			Volatile.Write(ref this._done, true);
			this.Schedule();
		}

		public override void OnError(Exception error)
		{
			this._error = error;
			Volatile.Write(ref this._done, true);
			this.Schedule();
		}

		public override void OnNext(TSource value)
		{
			this._queue.Enqueue(value);
			this.Schedule();
		}

		private void Schedule()
		{
			if (Volatile.Read(ref this._runDrainOnce) == 0 && Interlocked.CompareExchange(ref this._runDrainOnce, 1, 0) == 0)
			{
				Disposable.SetSingle(ref this._drainTask, this._scheduler.ScheduleLongRunning<ObserveOnObserverLongRunning<TSource>>(this, ObserveOnObserverLongRunning<TSource>.DrainLongRunning));
			}
			if (Interlocked.Increment(ref this._wip) == 1L)
			{
				object suspendGuard = this._suspendGuard;
				lock (suspendGuard)
				{
					Monitor.Pulse(this._suspendGuard);
				}
			}
		}

		protected override void Dispose(bool disposing)
		{
			Volatile.Write(ref this._disposed, true);
			object suspendGuard = this._suspendGuard;
			lock (suspendGuard)
			{
				Monitor.Pulse(this._suspendGuard);
			}
			Disposable.TryDispose(ref this._drainTask);
			base.Dispose(disposing);
		}

		private void Drain()
		{
			ConcurrentQueue<TSource> queue = this._queue;
			while (!Volatile.Read(ref this._disposed))
			{
				bool flag = Volatile.Read(ref this._done);
				TSource tsource;
				bool flag2 = queue.TryDequeue(out tsource);
				if (!flag || flag2)
				{
					if (flag2)
					{
						base.ForwardOnNext(tsource);
						if (Interlocked.Decrement(ref this._wip) != 0L)
						{
							continue;
						}
					}
					if (Volatile.Read(ref this._wip) == 0L && !Volatile.Read(ref this._disposed))
					{
						object suspendGuard = this._suspendGuard;
						if (Monitor.TryEnter(suspendGuard))
						{
							if (Volatile.Read(ref this._wip) == 0L && !Volatile.Read(ref this._disposed))
							{
								Monitor.Wait(suspendGuard);
							}
							Monitor.Exit(suspendGuard);
						}
					}
				}
				else
				{
					Exception error = this._error;
					if (error != null)
					{
						base.ForwardOnError(error);
						return;
					}
					base.ForwardOnCompleted();
					return;
				}
			}
			TSource tsource2;
			while (queue.TryDequeue(out tsource2))
			{
			}
		}

		private readonly ISchedulerLongRunning _scheduler;

		private readonly ConcurrentQueue<TSource> _queue;

		private readonly object _suspendGuard;

		private long _wip;

		private bool _done;

		private Exception _error;

		private bool _disposed;

		private int _runDrainOnce;

		private IDisposable _drainTask;

		private static readonly Action<ObserveOnObserverLongRunning<TSource>, ICancelable> DrainLongRunning = delegate(ObserveOnObserverLongRunning<TSource> self, ICancelable cancelable)
		{
			self.Drain();
		};
	}
}
