using System;
using System.Reactive.Disposables;
using System.Runtime.CompilerServices;
using System.Threading;

namespace System.Reactive.Subjects
{
	public sealed class AsyncSubject<T> : SubjectBase<T>, INotifyCompletion
	{
		public AsyncSubject()
		{
			this._observers = Array.Empty<AsyncSubject<T>.AsyncSubjectDisposable>();
		}

		public override bool HasObservers
		{
			get
			{
				return this._observers.Length != 0;
			}
		}

		public override bool IsDisposed
		{
			get
			{
				return Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers) == AsyncSubject<T>.Disposed;
			}
		}

		public override void OnCompleted()
		{
			for (;;)
			{
				AsyncSubject<T>.AsyncSubjectDisposable[] array = Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers);
				if (array == AsyncSubject<T>.Disposed)
				{
					break;
				}
				if (array == AsyncSubject<T>.Terminated)
				{
					return;
				}
				if (Interlocked.CompareExchange<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers, AsyncSubject<T>.Terminated, array) == array)
				{
					if (this._hasValue)
					{
						T value = this._value;
						foreach (AsyncSubject<T>.AsyncSubjectDisposable asyncSubjectDisposable in array)
						{
							if (!asyncSubjectDisposable.IsDisposed())
							{
								asyncSubjectDisposable.Downstream.OnNext(value);
								asyncSubjectDisposable.Downstream.OnCompleted();
							}
						}
					}
					else
					{
						foreach (AsyncSubject<T>.AsyncSubjectDisposable asyncSubjectDisposable2 in array)
						{
							if (!asyncSubjectDisposable2.IsDisposed())
							{
								asyncSubjectDisposable2.Downstream.OnCompleted();
							}
						}
					}
				}
			}
			this._exception = null;
			this.ThrowDisposed();
		}

		public override void OnError(Exception error)
		{
			if (error == null)
			{
				throw new ArgumentNullException("error");
			}
			for (;;)
			{
				AsyncSubject<T>.AsyncSubjectDisposable[] array = Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers);
				if (array == AsyncSubject<T>.Disposed)
				{
					break;
				}
				if (array == AsyncSubject<T>.Terminated)
				{
					return;
				}
				this._exception = error;
				if (Interlocked.CompareExchange<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers, AsyncSubject<T>.Terminated, array) == array)
				{
					foreach (AsyncSubject<T>.AsyncSubjectDisposable asyncSubjectDisposable in array)
					{
						if (!asyncSubjectDisposable.IsDisposed())
						{
							asyncSubjectDisposable.Downstream.OnError(error);
						}
					}
				}
			}
			this._exception = null;
			this._value = default(T);
			this.ThrowDisposed();
		}

		public override void OnNext(T value)
		{
			AsyncSubject<T>.AsyncSubjectDisposable[] array = Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers);
			if (array == AsyncSubject<T>.Disposed)
			{
				this._value = default(T);
				this._exception = null;
				this.ThrowDisposed();
				return;
			}
			if (array == AsyncSubject<T>.Terminated)
			{
				return;
			}
			this._value = value;
			this._hasValue = true;
		}

		public override IDisposable Subscribe(IObserver<T> observer)
		{
			if (observer == null)
			{
				throw new ArgumentNullException("observer");
			}
			AsyncSubject<T>.AsyncSubjectDisposable asyncSubjectDisposable = new AsyncSubject<T>.AsyncSubjectDisposable(this, observer);
			if (!this.Add(asyncSubjectDisposable))
			{
				Exception exception = this._exception;
				if (exception != null)
				{
					observer.OnError(exception);
				}
				else
				{
					if (this._hasValue)
					{
						observer.OnNext(this._value);
					}
					observer.OnCompleted();
				}
				return Disposable.Empty;
			}
			return asyncSubjectDisposable;
		}

		private bool Add(AsyncSubject<T>.AsyncSubjectDisposable inner)
		{
			for (;;)
			{
				AsyncSubject<T>.AsyncSubjectDisposable[] array = Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers);
				if (array == AsyncSubject<T>.Disposed)
				{
					break;
				}
				if (array == AsyncSubject<T>.Terminated)
				{
					return false;
				}
				int num = array.Length;
				AsyncSubject<T>.AsyncSubjectDisposable[] array2 = new AsyncSubject<T>.AsyncSubjectDisposable[num + 1];
				Array.Copy(array, 0, array2, 0, num);
				array2[num] = inner;
				if (Interlocked.CompareExchange<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers, array2, array) == array)
				{
					return true;
				}
			}
			this._value = default(T);
			this._exception = null;
			this.ThrowDisposed();
			return true;
		}

		private void Remove(AsyncSubject<T>.AsyncSubjectDisposable inner)
		{
			for (;;)
			{
				IL_0000:
				AsyncSubject<T>.AsyncSubjectDisposable[] array = Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers);
				int num = array.Length;
				if (num != 0)
				{
					int num2 = -1;
					int i = 0;
					while (i < num)
					{
						if (array[i] == inner)
						{
							num2 = i;
							IL_0035:
							if (num2 < 0)
							{
								return;
							}
							AsyncSubject<T>.AsyncSubjectDisposable[] array2;
							if (num == 1)
							{
								array2 = Array.Empty<AsyncSubject<T>.AsyncSubjectDisposable>();
							}
							else
							{
								array2 = new AsyncSubject<T>.AsyncSubjectDisposable[num - 1];
								Array.Copy(array, 0, array2, 0, num2);
								Array.Copy(array, num2 + 1, array2, num2, num - num2 - 1);
							}
							if (Interlocked.CompareExchange<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers, array2, array) == array)
							{
								return;
							}
							goto IL_0000;
						}
						else
						{
							i++;
						}
					}
					goto IL_0035;
				}
				break;
			}
		}

		private void ThrowDisposed()
		{
			throw new ObjectDisposedException(string.Empty);
		}

		public override void Dispose()
		{
			if (Interlocked.Exchange<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers, AsyncSubject<T>.Disposed) != AsyncSubject<T>.Disposed)
			{
				this._exception = null;
				this._value = default(T);
				this._hasValue = false;
			}
		}

		public AsyncSubject<T> GetAwaiter()
		{
			return this;
		}

		public void OnCompleted(Action continuation)
		{
			if (continuation == null)
			{
				throw new ArgumentNullException("continuation");
			}
			this.OnCompleted(continuation, true);
		}

		private void OnCompleted(Action continuation, bool originalContext)
		{
			this.Subscribe(new AsyncSubject<T>.AwaitObserver(continuation, originalContext));
		}

		public bool IsCompleted
		{
			get
			{
				return Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers) == AsyncSubject<T>.Terminated;
			}
		}

		public T GetResult()
		{
			if (Volatile.Read<AsyncSubject<T>.AsyncSubjectDisposable[]>(ref this._observers) != AsyncSubject<T>.Terminated)
			{
				ManualResetEvent e = new ManualResetEvent(false);
				this.OnCompleted(delegate
				{
					e.Set();
				}, false);
				e.WaitOne();
			}
			this._exception.ThrowIfNotNull();
			if (!this._hasValue)
			{
				throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
			}
			return this._value;
		}

		private AsyncSubject<T>.AsyncSubjectDisposable[] _observers;

		private T _value;

		private bool _hasValue;

		private Exception _exception;

		private static readonly AsyncSubject<T>.AsyncSubjectDisposable[] Terminated = new AsyncSubject<T>.AsyncSubjectDisposable[0];

		private static readonly AsyncSubject<T>.AsyncSubjectDisposable[] Disposed = new AsyncSubject<T>.AsyncSubjectDisposable[0];

		private sealed class AsyncSubjectDisposable : IDisposable
		{
			public AsyncSubjectDisposable(AsyncSubject<T> parent, IObserver<T> downstream)
			{
				this._parent = parent;
				this.Downstream = downstream;
			}

			public void Dispose()
			{
				AsyncSubject<T> asyncSubject = Interlocked.Exchange<AsyncSubject<T>>(ref this._parent, null);
				if (asyncSubject == null)
				{
					return;
				}
				asyncSubject.Remove(this);
			}

			internal bool IsDisposed()
			{
				return Volatile.Read<AsyncSubject<T>>(ref this._parent) == null;
			}

			internal readonly IObserver<T> Downstream;

			private AsyncSubject<T> _parent;
		}

		private sealed class AwaitObserver : IObserver<T>
		{
			public AwaitObserver(Action callback, bool originalContext)
			{
				if (originalContext)
				{
					this._context = SynchronizationContext.Current;
				}
				this._callback = callback;
			}

			public void OnCompleted()
			{
				this.InvokeOnOriginalContext();
			}

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

			public void OnNext(T value)
			{
			}

			private void InvokeOnOriginalContext()
			{
				if (this._context != null)
				{
					this._context.Post(delegate(object c)
					{
						((Action)c)();
					}, this._callback);
					return;
				}
				this._callback();
			}

			private readonly SynchronizationContext _context;

			private readonly Action _callback;
		}
	}
}
