using System;
using System.Collections.Concurrent;
using System.Threading;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class RedoSerializedObserver<X> : IObserver<X>
	{
		internal RedoSerializedObserver(IObserver<X> downstream)
		{
			this._downstream = downstream;
			this._queue = new ConcurrentQueue<X>();
		}

		public void OnCompleted()
		{
			if (Interlocked.CompareExchange<Exception>(ref this._terminalException, ExceptionHelper.Terminated, null) == null)
			{
				this.Drain();
			}
		}

		public void OnError(Exception error)
		{
			if (Interlocked.CompareExchange<Exception>(ref this._terminalException, error, null) == null)
			{
				this.Drain();
			}
		}

		public void OnNext(X value)
		{
			this._queue.Enqueue(value);
			this.Drain();
		}

		private void Clear()
		{
			X x;
			while (this._queue.TryDequeue(out x))
			{
			}
		}

		private void Drain()
		{
			if (Interlocked.Increment(ref this._wip) != 1)
			{
				return;
			}
			int num = 1;
			do
			{
				Exception ex = Volatile.Read<Exception>(ref this._terminalException);
				if (ex != null)
				{
					if (ex != RedoSerializedObserver<X>.SignaledIndicator)
					{
						Interlocked.Exchange<Exception>(ref this._terminalException, RedoSerializedObserver<X>.SignaledIndicator);
						if (ex != ExceptionHelper.Terminated)
						{
							this._downstream.OnError(ex);
						}
						else
						{
							this._downstream.OnCompleted();
						}
					}
					this.Clear();
				}
				else
				{
					X x;
					while (this._queue.TryDequeue(out x))
					{
						this._downstream.OnNext(x);
					}
				}
				num = Interlocked.Add(ref this._wip, -num);
			}
			while (num != 0);
		}

		private readonly IObserver<X> _downstream;

		private int _wip;

		private Exception _terminalException;

		private static readonly Exception SignaledIndicator = new Exception();

		private readonly ConcurrentQueue<X> _queue;
	}
}
