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 SelectMany<TSource, TResult>
	{
		internal class ObservableSelector : Producer<TResult, SelectMany<TSource, TResult>.ObservableSelector._>
		{
			public ObservableSelector(IObservable<TSource> source, Func<TSource, IObservable<TResult>> selector)
			{
				this._source = source;
				this._selector = selector;
			}

			protected override SelectMany<TSource, TResult>.ObservableSelector._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.ObservableSelector._(this, observer);
			}

			protected override void Run(SelectMany<TSource, TResult>.ObservableSelector._ sink)
			{
				sink.Run(this._source);
			}

			protected readonly IObservable<TSource> _source;

			protected readonly Func<TSource, IObservable<TResult>> _selector;

			internal class _ : Sink<TSource, TResult>
			{
				public _(SelectMany<TSource, TResult>.ObservableSelector parent, IObserver<TResult> observer)
					: base(observer)
				{
					this._selector = parent._selector;
				}

				public override void OnNext(TSource value)
				{
					IObservable<TResult> observable = null;
					try
					{
						observable = this._selector(value);
					}
					catch (Exception ex)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnError(ex);
						}
						return;
					}
					this.SubscribeInner(observable);
				}

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

				public override void OnCompleted()
				{
					this.Final();
				}

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

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

				protected void SubscribeInner(IObservable<TResult> inner)
				{
					SelectMany<TSource, TResult>.ObservableSelector._.InnerObserver innerObserver = new SelectMany<TSource, TResult>.ObservableSelector._.InnerObserver(this);
					this._group.Add(innerObserver);
					innerObserver.SetResource(inner.SubscribeSafe(innerObserver));
				}

				protected readonly object _gate = new object();

				private readonly Func<TSource, IObservable<TResult>> _selector;

				private readonly CompositeDisposable _group = new CompositeDisposable();

				private volatile bool _isStopped;

				private sealed class InnerObserver : SafeObserver<TResult>
				{
					public InnerObserver(SelectMany<TSource, TResult>.ObservableSelector._ parent)
					{
						this._parent = parent;
					}

					public override void OnNext(TResult 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 SelectMany<TSource, TResult>.ObservableSelector._ _parent;
				}
			}
		}

		internal sealed class ObservableSelectors : SelectMany<TSource, TResult>.ObservableSelector
		{
			public ObservableSelectors(IObservable<TSource> source, Func<TSource, IObservable<TResult>> selector, Func<Exception, IObservable<TResult>> selectorOnError, Func<IObservable<TResult>> selectorOnCompleted)
				: base(source, selector)
			{
				this._selectorOnError = selectorOnError;
				this._selectorOnCompleted = selectorOnCompleted;
			}

			protected override SelectMany<TSource, TResult>.ObservableSelector._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.ObservableSelectors._(this, observer);
			}

			private readonly Func<Exception, IObservable<TResult>> _selectorOnError;

			private readonly Func<IObservable<TResult>> _selectorOnCompleted;

			internal new sealed class _ : SelectMany<TSource, TResult>.ObservableSelector._
			{
				public _(SelectMany<TSource, TResult>.ObservableSelectors parent, IObserver<TResult> observer)
					: base(parent, observer)
				{
					this._selectorOnError = parent._selectorOnError;
					this._selectorOnCompleted = parent._selectorOnCompleted;
				}

				public override void OnError(Exception error)
				{
					if (this._selectorOnError != null)
					{
						IObservable<TResult> observable = null;
						try
						{
							observable = this._selectorOnError(error);
						}
						catch (Exception ex)
						{
							object gate = this._gate;
							lock (gate)
							{
								base.ForwardOnError(ex);
							}
							return;
						}
						base.SubscribeInner(observable);
						base.Final();
						return;
					}
					base.OnError(error);
				}

				public override void OnCompleted()
				{
					if (this._selectorOnCompleted != null)
					{
						IObservable<TResult> observable = null;
						try
						{
							observable = this._selectorOnCompleted();
						}
						catch (Exception ex)
						{
							object gate = this._gate;
							lock (gate)
							{
								base.ForwardOnError(ex);
							}
							return;
						}
						base.SubscribeInner(observable);
					}
					base.Final();
				}

				private readonly Func<Exception, IObservable<TResult>> _selectorOnError;

				private readonly Func<IObservable<TResult>> _selectorOnCompleted;
			}
		}

		internal class ObservableSelectorIndexed : Producer<TResult, SelectMany<TSource, TResult>.ObservableSelectorIndexed._>
		{
			public ObservableSelectorIndexed(IObservable<TSource> source, Func<TSource, int, IObservable<TResult>> selector)
			{
				this._source = source;
				this._selector = selector;
			}

			protected override SelectMany<TSource, TResult>.ObservableSelectorIndexed._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.ObservableSelectorIndexed._(this, observer);
			}

			protected override void Run(SelectMany<TSource, TResult>.ObservableSelectorIndexed._ sink)
			{
				sink.Run(this._source);
			}

			protected readonly IObservable<TSource> _source;

			protected readonly Func<TSource, int, IObservable<TResult>> _selector;

			internal class _ : Sink<TSource, TResult>
			{
				public _(SelectMany<TSource, TResult>.ObservableSelectorIndexed parent, IObserver<TResult> observer)
					: base(observer)
				{
					this._selector = parent._selector;
				}

				public override void OnNext(TSource value)
				{
					IObservable<TResult> observable = null;
					try
					{
						Func<TSource, int, IObservable<TResult>> selector = this._selector;
						int index = this._index;
						this._index = checked(index + 1);
						observable = selector(value, index);
					}
					catch (Exception ex)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnError(ex);
						}
						return;
					}
					this.SubscribeInner(observable);
				}

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

				public override void OnCompleted()
				{
					this.Final();
				}

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

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

				protected void SubscribeInner(IObservable<TResult> inner)
				{
					SelectMany<TSource, TResult>.ObservableSelectorIndexed._.InnerObserver innerObserver = new SelectMany<TSource, TResult>.ObservableSelectorIndexed._.InnerObserver(this);
					this._group.Add(innerObserver);
					innerObserver.SetResource(inner.SubscribeSafe(innerObserver));
				}

				private readonly object _gate = new object();

				private readonly CompositeDisposable _group = new CompositeDisposable();

				protected readonly Func<TSource, int, IObservable<TResult>> _selector;

				private int _index;

				private volatile bool _isStopped;

				private sealed class InnerObserver : SafeObserver<TResult>
				{
					public InnerObserver(SelectMany<TSource, TResult>.ObservableSelectorIndexed._ parent)
					{
						this._parent = parent;
					}

					public override void OnNext(TResult 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 SelectMany<TSource, TResult>.ObservableSelectorIndexed._ _parent;
				}
			}
		}

		internal sealed class ObservableSelectorsIndexed : SelectMany<TSource, TResult>.ObservableSelectorIndexed
		{
			public ObservableSelectorsIndexed(IObservable<TSource> source, Func<TSource, int, IObservable<TResult>> selector, Func<Exception, IObservable<TResult>> selectorOnError, Func<IObservable<TResult>> selectorOnCompleted)
				: base(source, selector)
			{
				this._selectorOnError = selectorOnError;
				this._selectorOnCompleted = selectorOnCompleted;
			}

			protected override SelectMany<TSource, TResult>.ObservableSelectorIndexed._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.ObservableSelectorsIndexed._(this, observer);
			}

			private readonly Func<Exception, IObservable<TResult>> _selectorOnError;

			private readonly Func<IObservable<TResult>> _selectorOnCompleted;

			internal new sealed class _ : SelectMany<TSource, TResult>.ObservableSelectorIndexed._
			{
				public _(SelectMany<TSource, TResult>.ObservableSelectorsIndexed parent, IObserver<TResult> observer)
					: base(parent, observer)
				{
					this._selectorOnError = parent._selectorOnError;
					this._selectorOnCompleted = parent._selectorOnCompleted;
				}

				public override void OnError(Exception error)
				{
					if (this._selectorOnError != null)
					{
						IObservable<TResult> observable = null;
						try
						{
							observable = this._selectorOnError(error);
						}
						catch (Exception ex)
						{
							object gate = this._gate;
							lock (gate)
							{
								base.ForwardOnError(ex);
							}
							return;
						}
						base.SubscribeInner(observable);
						base.Final();
						return;
					}
					base.OnError(error);
				}

				public override void OnCompleted()
				{
					if (this._selectorOnCompleted != null)
					{
						IObservable<TResult> observable = null;
						try
						{
							observable = this._selectorOnCompleted();
						}
						catch (Exception ex)
						{
							object gate = this._gate;
							lock (gate)
							{
								base.ForwardOnError(ex);
							}
							return;
						}
						base.SubscribeInner(observable);
					}
					base.Final();
				}

				private readonly object _gate = new object();

				private readonly Func<Exception, IObservable<TResult>> _selectorOnError;

				private readonly Func<IObservable<TResult>> _selectorOnCompleted;
			}
		}

		internal sealed class EnumerableSelector : Producer<TResult, SelectMany<TSource, TResult>.EnumerableSelector._>
		{
			public EnumerableSelector(IObservable<TSource> source, Func<TSource, IEnumerable<TResult>> selector)
			{
				this._source = source;
				this._selector = selector;
			}

			protected override SelectMany<TSource, TResult>.EnumerableSelector._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.EnumerableSelector._(this, observer);
			}

			protected override void Run(SelectMany<TSource, TResult>.EnumerableSelector._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly Func<TSource, IEnumerable<TResult>> _selector;

			internal sealed class _ : Sink<TSource, TResult>
			{
				public _(SelectMany<TSource, TResult>.EnumerableSelector parent, IObserver<TResult> observer)
					: base(observer)
				{
					this._selector = parent._selector;
				}

				public override void OnNext(TSource value)
				{
					IEnumerable<TResult> enumerable = null;
					try
					{
						enumerable = this._selector(value);
					}
					catch (Exception ex)
					{
						base.ForwardOnError(ex);
						return;
					}
					IEnumerator<TResult> enumerator;
					try
					{
						enumerator = enumerable.GetEnumerator();
					}
					catch (Exception ex2)
					{
						base.ForwardOnError(ex2);
						return;
					}
					using (enumerator)
					{
						bool flag = true;
						while (flag)
						{
							TResult tresult = default(TResult);
							try
							{
								if (flag = enumerator.MoveNext())
								{
									tresult = enumerator.Current;
								}
							}
							catch (Exception ex3)
							{
								base.ForwardOnError(ex3);
								break;
							}
							if (flag)
							{
								base.ForwardOnNext(tresult);
							}
						}
					}
				}

				private readonly Func<TSource, IEnumerable<TResult>> _selector;
			}
		}

		internal sealed class EnumerableSelectorIndexed : Producer<TResult, SelectMany<TSource, TResult>.EnumerableSelectorIndexed._>
		{
			public EnumerableSelectorIndexed(IObservable<TSource> source, Func<TSource, int, IEnumerable<TResult>> selector)
			{
				this._source = source;
				this._selector = selector;
			}

			protected override SelectMany<TSource, TResult>.EnumerableSelectorIndexed._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.EnumerableSelectorIndexed._(this, observer);
			}

			protected override void Run(SelectMany<TSource, TResult>.EnumerableSelectorIndexed._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly Func<TSource, int, IEnumerable<TResult>> _selector;

			internal sealed class _ : Sink<TSource, TResult>
			{
				public _(SelectMany<TSource, TResult>.EnumerableSelectorIndexed parent, IObserver<TResult> observer)
					: base(observer)
				{
					this._selector = parent._selector;
				}

				public override void OnNext(TSource value)
				{
					IEnumerable<TResult> enumerable = null;
					try
					{
						Func<TSource, int, IEnumerable<TResult>> selector = this._selector;
						int index = this._index;
						this._index = checked(index + 1);
						enumerable = selector(value, index);
					}
					catch (Exception ex)
					{
						base.ForwardOnError(ex);
						return;
					}
					IEnumerator<TResult> enumerator;
					try
					{
						enumerator = enumerable.GetEnumerator();
					}
					catch (Exception ex2)
					{
						base.ForwardOnError(ex2);
						return;
					}
					using (enumerator)
					{
						bool flag = true;
						while (flag)
						{
							TResult tresult = default(TResult);
							try
							{
								if (flag = enumerator.MoveNext())
								{
									tresult = enumerator.Current;
								}
							}
							catch (Exception ex3)
							{
								base.ForwardOnError(ex3);
								break;
							}
							if (flag)
							{
								base.ForwardOnNext(tresult);
							}
						}
					}
				}

				private readonly Func<TSource, int, IEnumerable<TResult>> _selector;

				private int _index;
			}
		}

		internal sealed class TaskSelector : Producer<TResult, SelectMany<TSource, TResult>.TaskSelector._>
		{
			public TaskSelector(IObservable<TSource> source, Func<TSource, CancellationToken, Task<TResult>> selector)
			{
				this._source = source;
				this._selector = selector;
			}

			protected override SelectMany<TSource, TResult>.TaskSelector._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.TaskSelector._(this, observer);
			}

			protected override void Run(SelectMany<TSource, TResult>.TaskSelector._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly Func<TSource, CancellationToken, Task<TResult>> _selector;

			internal sealed class _ : Sink<TSource, TResult>
			{
				public _(SelectMany<TSource, TResult>.TaskSelector parent, IObserver<TResult> observer)
					: base(observer)
				{
					this._selector = parent._selector;
				}

				public override void Run(IObservable<TSource> source)
				{
					this._count = 1;
					base.Run(source);
				}

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

				public override void OnNext(TSource value)
				{
					Task<TResult> task = null;
					try
					{
						Interlocked.Increment(ref this._count);
						task = this._selector(value, this._cts.Token);
					}
					catch (Exception ex)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnError(ex);
						}
						return;
					}
					if (task.IsCompleted)
					{
						this.OnCompletedTask(task);
						return;
					}
					task.ContinueWith(delegate(Task<TResult> closureTask, object thisObject)
					{
						((SelectMany<TSource, TResult>.TaskSelector._)thisObject).OnCompletedTask(closureTask);
					}, this, this._cts.Token);
				}

				private void OnCompletedTask(Task<TResult> task)
				{
					object obj;
					switch (task.Status)
					{
					case TaskStatus.RanToCompletion:
						obj = this._gate;
						lock (obj)
						{
							base.ForwardOnNext(task.Result);
						}
						this.OnCompleted();
						return;
					case TaskStatus.Canceled:
						if (this._cts.IsCancellationRequested)
						{
							return;
						}
						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();
						}
					}
				}

				private readonly object _gate = new object();

				private readonly CancellationTokenSource _cts = new CancellationTokenSource();

				private readonly Func<TSource, CancellationToken, Task<TResult>> _selector;

				private volatile int _count;
			}
		}

		internal sealed class TaskSelectorIndexed : Producer<TResult, SelectMany<TSource, TResult>.TaskSelectorIndexed._>
		{
			public TaskSelectorIndexed(IObservable<TSource> source, Func<TSource, int, CancellationToken, Task<TResult>> selector)
			{
				this._source = source;
				this._selector = selector;
			}

			protected override SelectMany<TSource, TResult>.TaskSelectorIndexed._ CreateSink(IObserver<TResult> observer)
			{
				return new SelectMany<TSource, TResult>.TaskSelectorIndexed._(this, observer);
			}

			protected override void Run(SelectMany<TSource, TResult>.TaskSelectorIndexed._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly Func<TSource, int, CancellationToken, Task<TResult>> _selector;

			internal sealed class _ : Sink<TSource, TResult>
			{
				public _(SelectMany<TSource, TResult>.TaskSelectorIndexed parent, IObserver<TResult> observer)
					: base(observer)
				{
					this._selector = parent._selector;
				}

				public override void Run(IObservable<TSource> source)
				{
					this._count = 1;
					base.Run(source);
				}

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

				public override void OnNext(TSource value)
				{
					Task<TResult> task = null;
					try
					{
						Interlocked.Increment(ref this._count);
						Func<TSource, int, CancellationToken, Task<TResult>> selector = this._selector;
						int index = this._index;
						this._index = checked(index + 1);
						task = selector(value, index, this._cts.Token);
					}
					catch (Exception ex)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnError(ex);
						}
						return;
					}
					if (task.IsCompleted)
					{
						this.OnCompletedTask(task);
						return;
					}
					task.ContinueWith(delegate(Task<TResult> closureTask, object thisObject)
					{
						((SelectMany<TSource, TResult>.TaskSelectorIndexed._)thisObject).OnCompletedTask(closureTask);
					}, this, this._cts.Token);
				}

				private void OnCompletedTask(Task<TResult> task)
				{
					object obj;
					switch (task.Status)
					{
					case TaskStatus.RanToCompletion:
						obj = this._gate;
						lock (obj)
						{
							base.ForwardOnNext(task.Result);
						}
						this.OnCompleted();
						return;
					case TaskStatus.Canceled:
						if (this._cts.IsCancellationRequested)
						{
							return;
						}
						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();
						}
					}
				}

				private readonly object _gate = new object();

				private readonly CancellationTokenSource _cts = new CancellationTokenSource();

				private readonly Func<TSource, int, CancellationToken, Task<TResult>> _selector;

				private volatile int _count;

				private int _index;
			}
		}
	}
}
