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

namespace System.Reactive
{
	internal sealed class ObserveOnObserverNew<T> : IdentitySink<T>
	{
		public ObserveOnObserverNew(IScheduler scheduler, IObserver<T> downstream)
			: base(downstream)
		{
			this._scheduler = scheduler;
			this._queue = new ConcurrentQueue<T>();
		}

		protected override void Dispose(bool disposing)
		{
			Volatile.Write(ref this._disposed, true);
			base.Dispose(disposing);
			if (disposing)
			{
				Disposable.TryDispose(ref this._task);
				this.Clear(this._queue);
			}
		}

		private void Clear(ConcurrentQueue<T> q)
		{
			T t;
			while (q.TryDequeue(out t))
			{
			}
		}

		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(T value)
		{
			this._queue.Enqueue(value);
			this.Schedule();
		}

		private void Schedule()
		{
			if (Interlocked.Increment(ref this._wip) == 1)
			{
				SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable();
				if (Disposable.TrySetMultiple(ref this._task, singleAssignmentDisposable))
				{
					singleAssignmentDisposable.Disposable = this._scheduler.Schedule<ObserveOnObserverNew<T>>(this, ObserveOnObserverNew<T>.DrainShortRunningFunc);
				}
				if (Volatile.Read(ref this._disposed))
				{
					this.Clear(this._queue);
				}
			}
		}

		private IDisposable DrainShortRunning(IScheduler recursiveScheduler)
		{
			this.DrainStep(this._queue);
			if (Interlocked.Decrement(ref this._wip) != 0)
			{
				IDisposable disposable = recursiveScheduler.Schedule<ObserveOnObserverNew<T>>(this, ObserveOnObserverNew<T>.DrainShortRunningFunc);
				Disposable.TrySetMultiple(ref this._task, disposable);
			}
			return Disposable.Empty;
		}

		private void DrainStep(ConcurrentQueue<T> q)
		{
			if (Volatile.Read(ref this._disposed))
			{
				this.Clear(q);
				return;
			}
			bool flag;
			if (flag = Volatile.Read(ref this._done))
			{
				Exception error = this._error;
				if (error != null)
				{
					Volatile.Write(ref this._disposed, true);
					base.ForwardOnError(error);
					return;
				}
			}
			T t;
			bool flag2 = !q.TryDequeue(out t);
			if (flag && flag2)
			{
				Volatile.Write(ref this._disposed, true);
				base.ForwardOnCompleted();
				return;
			}
			if (flag2)
			{
				return;
			}
			base.ForwardOnNext(t);
		}

		private readonly IScheduler _scheduler;

		private readonly ConcurrentQueue<T> _queue;

		private IDisposable _task;

		private int _wip;

		private bool _done;

		private Exception _error;

		private bool _disposed;

		private static readonly Func<IScheduler, ObserveOnObserverNew<T>, IDisposable> DrainShortRunningFunc = (IScheduler scheduler, ObserveOnObserverNew<T> self) => self.DrainShortRunning(scheduler);
	}
}
