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

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class ConcatMany<T> : IObservable<T>
	{
		internal ConcatMany(IObservable<IObservable<T>> sources)
		{
			this._sources = sources;
		}

		public IDisposable Subscribe(IObserver<T> observer)
		{
			if (observer == null)
			{
				throw new ArgumentNullException("observer");
			}
			ConcatMany<T>.ConcatManyOuterObserver concatManyOuterObserver = new ConcatMany<T>.ConcatManyOuterObserver(observer);
			IDisposable disposable = this._sources.SubscribeSafe(concatManyOuterObserver);
			concatManyOuterObserver.OnSubscribe(disposable);
			return concatManyOuterObserver;
		}

		private readonly IObservable<IObservable<T>> _sources;

		internal sealed class ConcatManyOuterObserver : IObserver<IObservable<T>>, IDisposable
		{
			internal ConcatManyOuterObserver(IObserver<T> downstream)
			{
				this._downstream = downstream;
				this._queue = new ConcurrentQueue<IObservable<T>>();
				this._innerObserver = new ConcatMany<T>.ConcatManyOuterObserver.InnerObserver(this);
			}

			internal void OnSubscribe(IDisposable d)
			{
				Disposable.SetSingle(ref this._upstream, d);
			}

			public void Dispose()
			{
				this._innerObserver.Dispose();
				this.DisposeMain();
			}

			private void DisposeMain()
			{
				Disposable.TryDispose(ref this._upstream);
			}

			private bool IsDisposed()
			{
				return Disposable.GetIsDisposed(ref this._upstream);
			}

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

			public void OnError(Exception error)
			{
				if (Interlocked.CompareExchange<Exception>(ref this._error, error, null) == null)
				{
					Volatile.Write(ref this._done, true);
					this.Drain();
				}
			}

			public void OnNext(IObservable<T> value)
			{
				this._queue.Enqueue(value);
				this.Drain();
			}

			private void InnerNext(T item)
			{
				this._downstream.OnNext(item);
			}

			private void InnerError(Exception error)
			{
				if (this._innerObserver.Finish() && Interlocked.CompareExchange<Exception>(ref this._error, error, null) == null)
				{
					Volatile.Write(ref this._done, true);
					Volatile.Write(ref this._active, 0);
					this.Drain();
				}
			}

			private void InnerComplete()
			{
				if (this._innerObserver.Finish())
				{
					Volatile.Write(ref this._active, 0);
					this.Drain();
				}
			}

			private void Drain()
			{
				if (Interlocked.Increment(ref this._trampoline) != 1)
				{
					return;
				}
				do
				{
					if (this.IsDisposed())
					{
						IObservable<T> observable;
						while (this._queue.TryDequeue(out observable))
						{
						}
					}
					else if (Volatile.Read(ref this._active) == 0)
					{
						bool flag;
						if (flag = Volatile.Read(ref this._done))
						{
							Exception ex = Volatile.Read<Exception>(ref this._error);
							if (ex != null)
							{
								this._downstream.OnError(ex);
								this.DisposeMain();
								goto IL_00AF;
							}
						}
						IObservable<T> observable2;
						if (this._queue.TryDequeue(out observable2))
						{
							SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable();
							if (this._innerObserver.SetDisposable(singleAssignmentDisposable))
							{
								Interlocked.Exchange(ref this._active, 1);
								singleAssignmentDisposable.Disposable = observable2.SubscribeSafe(this._innerObserver);
							}
						}
						else if (flag)
						{
							this._downstream.OnCompleted();
							this.DisposeMain();
						}
					}
					IL_00AF:;
				}
				while (Interlocked.Decrement(ref this._trampoline) != 0);
			}

			private readonly IObserver<T> _downstream;

			private readonly ConcurrentQueue<IObservable<T>> _queue;

			private readonly ConcatMany<T>.ConcatManyOuterObserver.InnerObserver _innerObserver;

			private IDisposable _upstream;

			private int _trampoline;

			private Exception _error;

			private bool _done;

			private int _active;

			internal sealed class InnerObserver : IObserver<T>, IDisposable
			{
				internal InnerObserver(ConcatMany<T>.ConcatManyOuterObserver parent)
				{
					this._parent = parent;
				}

				internal bool SetDisposable(SingleAssignmentDisposable sad)
				{
					return Disposable.TrySetSingle(ref this.Upstream, sad) == TrySetSingleResult.Success;
				}

				internal bool Finish()
				{
					IDisposable disposable = Volatile.Read<IDisposable>(ref this.Upstream);
					if (disposable != BooleanDisposable.True && Interlocked.CompareExchange<IDisposable>(ref this.Upstream, null, disposable) == disposable)
					{
						disposable.Dispose();
						return true;
					}
					return false;
				}

				public void Dispose()
				{
					Disposable.TryDispose(ref this.Upstream);
				}

				public void OnCompleted()
				{
					this._parent.InnerComplete();
				}

				public void OnError(Exception error)
				{
					this._parent.InnerError(error);
				}

				public void OnNext(T value)
				{
					this._parent.InnerNext(value);
				}

				private readonly ConcatMany<T>.ConcatManyOuterObserver _parent;

				internal IDisposable Upstream;
			}
		}
	}
}
