using System;
using System.Collections.Generic;
using System.Reactive.Disposables;
using System.Threading;
using System.Threading.Tasks;

namespace System.Reactive.Linq.ObservableImpl
{
	internal static class Merge<TSource>
	{
		internal sealed class ObservablesMaxConcurrency : Producer<TSource, Merge<TSource>.ObservablesMaxConcurrency._>
		{
			public ObservablesMaxConcurrency(IObservable<IObservable<TSource>> sources, int maxConcurrent)
			{
				this._sources = sources;
				this._maxConcurrent = maxConcurrent;
			}

			protected override Merge<TSource>.ObservablesMaxConcurrency._ CreateSink(IObserver<TSource> observer)
			{
				return new Merge<TSource>.ObservablesMaxConcurrency._(this._maxConcurrent, observer);
			}

			protected override void Run(Merge<TSource>.ObservablesMaxConcurrency._ sink)
			{
				sink.Run(this._sources);
			}

			private readonly IObservable<IObservable<TSource>> _sources;

			private readonly int _maxConcurrent;

			internal sealed class _ : Sink<IObservable<TSource>, TSource>
			{
				public _(int maxConcurrent, IObserver<TSource> observer)
					: base(observer)
				{
					this._maxConcurrent = maxConcurrent;
				}

				public override void OnNext(IObservable<TSource> value)
				{
					object gate = this._gate;
					lock (gate)
					{
						if (this._activeCount < this._maxConcurrent)
						{
							this._activeCount++;
							this.Subscribe(value);
						}
						else
						{
							this._q.Enqueue(value);
						}
					}
				}

				public override void OnError(Exception error)
				{
					object gate = this._gate;
					lock (gate)
					{
						base.ForwardOnError(error);
					}
				}

				public override void OnCompleted()
				{
					object gate = this._gate;
					lock (gate)
					{
						this._isStopped = true;
						if (this._activeCount == 0)
						{
							base.ForwardOnCompleted();
						}
						else
						{
							base.DisposeUpstream();
						}
					}
				}

				protected override void Dispose(bool disposing)
				{
					base.Dispose(disposing);
					if (disposing)
					{
						this._group.Dispose();
					}
				}

				private void Subscribe(IObservable<TSource> innerSource)
				{
					Merge<TSource>.ObservablesMaxConcurrency._.InnerObserver innerObserver = new Merge<TSource>.ObservablesMaxConcurrency._.InnerObserver(this);
					this._group.Add(innerObserver);
					innerObserver.SetResource(innerSource.SubscribeSafe(innerObserver));
				}

				private readonly int _maxConcurrent;

				private readonly object _gate = new object();

				private readonly Queue<IObservable<TSource>> _q = new Queue<IObservable<TSource>>();

				private volatile bool _isStopped;

				private readonly CompositeDisposable _group = new CompositeDisposable();

				private int _activeCount;

				private sealed class InnerObserver : SafeObserver<TSource>
				{
					public InnerObserver(Merge<TSource>.ObservablesMaxConcurrency._ parent)
					{
						this._parent = parent;
					}

					public override void OnNext(TSource value)
					{
						object gate = this._parent._gate;
						lock (gate)
						{
							this._parent.ForwardOnNext(value);
						}
					}

					public override void OnError(Exception error)
					{
						object gate = this._parent._gate;
						lock (gate)
						{
							this._parent.ForwardOnError(error);
						}
					}

					public override void OnCompleted()
					{
						this._parent._group.Remove(this);
						object gate = this._parent._gate;
						lock (gate)
						{
							if (this._parent._q.Count > 0)
							{
								IObservable<TSource> observable = this._parent._q.Dequeue();
								this._parent.Subscribe(observable);
							}
							else
							{
								this._parent._activeCount--;
								if (this._parent._isStopped && this._parent._activeCount == 0)
								{
									this._parent.ForwardOnCompleted();
								}
							}
						}
					}

					private readonly Merge<TSource>.ObservablesMaxConcurrency._ _parent;
				}
			}
		}

		internal sealed class Observables : Producer<TSource, Merge<TSource>.Observables._>
		{
			public Observables(IObservable<IObservable<TSource>> sources)
			{
				this._sources = sources;
			}

			protected override Merge<TSource>.Observables._ CreateSink(IObserver<TSource> observer)
			{
				return new Merge<TSource>.Observables._(observer);
			}

			protected override void Run(Merge<TSource>.Observables._ sink)
			{
				sink.Run(this._sources);
			}

			private readonly IObservable<IObservable<TSource>> _sources;

			internal sealed class _ : Sink<IObservable<TSource>, TSource>
			{
				public _(IObserver<TSource> observer)
					: base(observer)
				{
				}

				public override void OnNext(IObservable<TSource> value)
				{
					Merge<TSource>.Observables._.InnerObserver innerObserver = new Merge<TSource>.Observables._.InnerObserver(this);
					this._group.Add(innerObserver);
					innerObserver.SetResource(value.SubscribeSafe(innerObserver));
				}

				public override void OnError(Exception error)
				{
					object gate = this._gate;
					lock (gate)
					{
						base.ForwardOnError(error);
					}
				}

				public override void OnCompleted()
				{
					this._isStopped = true;
					if (this._group.Count == 0)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnCompleted();
							return;
						}
					}
					base.DisposeUpstream();
				}

				protected override void Dispose(bool disposing)
				{
					base.Dispose(disposing);
					if (disposing)
					{
						this._group.Dispose();
					}
				}

				private readonly object _gate = new object();

				private volatile bool _isStopped;

				private readonly CompositeDisposable _group = new CompositeDisposable();

				private sealed class InnerObserver : SafeObserver<TSource>
				{
					public InnerObserver(Merge<TSource>.Observables._ parent)
					{
						this._parent = parent;
					}

					public override void OnNext(TSource value)
					{
						object gate = this._parent._gate;
						lock (gate)
						{
							this._parent.ForwardOnNext(value);
						}
					}

					public override void OnError(Exception error)
					{
						object gate = this._parent._gate;
						lock (gate)
						{
							this._parent.ForwardOnError(error);
						}
					}

					public override void OnCompleted()
					{
						this._parent._group.Remove(this);
						if (this._parent._isStopped && this._parent._group.Count == 0)
						{
							object gate = this._parent._gate;
							lock (gate)
							{
								this._parent.ForwardOnCompleted();
							}
						}
					}

					private readonly Merge<TSource>.Observables._ _parent;
				}
			}
		}

		internal sealed class Tasks : Producer<TSource, Merge<TSource>.Tasks._>
		{
			public Tasks(IObservable<Task<TSource>> sources)
			{
				this._sources = sources;
			}

			protected override Merge<TSource>.Tasks._ CreateSink(IObserver<TSource> observer)
			{
				return new Merge<TSource>.Tasks._(observer);
			}

			protected override void Run(Merge<TSource>.Tasks._ sink)
			{
				sink.Run(this._sources);
			}

			private readonly IObservable<Task<TSource>> _sources;

			internal sealed class _ : Sink<Task<TSource>, TSource>
			{
				public _(IObserver<TSource> observer)
					: base(observer)
				{
				}

				public override void OnNext(Task<TSource> value)
				{
					Interlocked.Increment(ref this._count);
					if (value.IsCompleted)
					{
						this.OnCompletedTask(value);
						return;
					}
					value.ContinueWith(delegate(Task<TSource> t, object thisObject)
					{
						((Merge<TSource>.Tasks._)thisObject).OnCompletedTask(t);
					}, this, this._cts.Token);
				}

				private void OnCompletedTask(Task<TSource> task)
				{
					object obj;
					switch (task.Status)
					{
					case TaskStatus.RanToCompletion:
						obj = this._gate;
						lock (obj)
						{
							base.ForwardOnNext(task.Result);
						}
						this.OnCompleted();
						return;
					case TaskStatus.Canceled:
						obj = this._gate;
						lock (obj)
						{
							base.ForwardOnError(new TaskCanceledException(task));
							return;
						}
						break;
					case TaskStatus.Faulted:
						break;
					default:
						return;
					}
					obj = this._gate;
					lock (obj)
					{
						base.ForwardOnError(task.Exception.InnerException);
					}
				}

				public override void OnError(Exception error)
				{
					object gate = this._gate;
					lock (gate)
					{
						base.ForwardOnError(error);
					}
				}

				public override void OnCompleted()
				{
					if (Interlocked.Decrement(ref this._count) == 0)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnCompleted();
						}
					}
				}

				protected override void Dispose(bool disposing)
				{
					if (disposing)
					{
						this._cts.Cancel();
					}
					base.Dispose(disposing);
				}

				private readonly object _gate = new object();

				private readonly CancellationTokenSource _cts = new CancellationTokenSource();

				private volatile int _count = 1;
			}
		}
	}
}
