using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Subjects;
using System.Runtime.CompilerServices;

namespace System.Reactive.Linq
{
	internal class QueryLanguageEx : IQueryLanguageEx
	{
		public virtual IObservable<TResult> Create<TResult>(Func<IObserver<TResult>, IEnumerable<IObservable<object>>> iteratorMethod)
		{
			return new QueryLanguageEx.CreateWithEnumerableObservable<TResult>(iteratorMethod);
		}

		public virtual IObservable<Unit> Create(Func<IEnumerable<IObservable<object>>> iteratorMethod)
		{
			return new QueryLanguageEx.CreateWithOnlyEnumerableObservable<Unit>(iteratorMethod);
		}

		public virtual IObservable<TSource> Expand<TSource>(IObservable<TSource> source, Func<TSource, IObservable<TSource>> selector, IScheduler scheduler)
		{
			return new QueryLanguageEx.ExpandObservable<TSource>(source, selector, scheduler);
		}

		public virtual IObservable<TSource> Expand<TSource>(IObservable<TSource> source, Func<TSource, IObservable<TSource>> selector)
		{
			return source.Expand(selector, SchedulerDefaults.Iteration);
		}

		public virtual IObservable<TResult> ForkJoin<TFirst, TSecond, TResult>(IObservable<TFirst> first, IObservable<TSecond> second, Func<TFirst, TSecond, TResult> resultSelector)
		{
			return QueryLanguageEx.Combine<TFirst, TSecond, TResult>(first, second, delegate(IObserver<TResult> observer, IDisposable leftSubscription, IDisposable rightSubscription)
			{
				bool leftStopped = false;
				bool rightStopped = false;
				bool hasLeft = false;
				bool hasRight = false;
				TFirst lastLeft = default(TFirst);
				TSecond lastRight = default(TSecond);
				return new BinaryObserver<TFirst, TSecond>(delegate(Notification<TFirst> left)
				{
					switch (left.Kind)
					{
					case NotificationKind.OnNext:
						hasLeft = true;
						lastLeft = left.Value;
						return;
					case NotificationKind.OnError:
						rightSubscription.Dispose();
						observer.OnError(left.Exception);
						return;
					case NotificationKind.OnCompleted:
						leftStopped = true;
						if (rightStopped)
						{
							if (!hasLeft)
							{
								observer.OnCompleted();
								return;
							}
							if (!hasRight)
							{
								observer.OnCompleted();
								return;
							}
							TResult tresult;
							try
							{
								tresult = resultSelector(lastLeft, lastRight);
							}
							catch (Exception ex)
							{
								observer.OnError(ex);
								return;
							}
							observer.OnNext(tresult);
							observer.OnCompleted();
						}
						return;
					default:
						return;
					}
				}, delegate(Notification<TSecond> right)
				{
					switch (right.Kind)
					{
					case NotificationKind.OnNext:
						hasRight = true;
						lastRight = right.Value;
						return;
					case NotificationKind.OnError:
						leftSubscription.Dispose();
						observer.OnError(right.Exception);
						return;
					case NotificationKind.OnCompleted:
						rightStopped = true;
						if (leftStopped)
						{
							if (!hasLeft)
							{
								observer.OnCompleted();
								return;
							}
							if (!hasRight)
							{
								observer.OnCompleted();
								return;
							}
							TResult tresult2;
							try
							{
								tresult2 = resultSelector(lastLeft, lastRight);
							}
							catch (Exception ex2)
							{
								observer.OnError(ex2);
								return;
							}
							observer.OnNext(tresult2);
							observer.OnCompleted();
						}
						return;
					default:
						return;
					}
				});
			});
		}

		public virtual IObservable<TSource[]> ForkJoin<TSource>(params IObservable<TSource>[] sources)
		{
			return sources.ForkJoin<TSource>();
		}

		public virtual IObservable<TSource[]> ForkJoin<TSource>(IEnumerable<IObservable<TSource>> sources)
		{
			return new QueryLanguageEx.ForkJoinObservable<TSource>(sources);
		}

		public virtual IObservable<TResult> Let<TSource, TResult>(IObservable<TSource> source, Func<IObservable<TSource>, IObservable<TResult>> function)
		{
			return function(source);
		}

		public virtual IObservable<TResult> ManySelect<TSource, TResult>(IObservable<TSource> source, Func<IObservable<TSource>, TResult> selector)
		{
			return this.ManySelect<TSource, TResult>(source, selector, DefaultScheduler.Instance);
		}

		public virtual IObservable<TResult> ManySelect<TSource, TResult>(IObservable<TSource> source, Func<IObservable<TSource>, TResult> selector, IScheduler scheduler)
		{
			return Observable.Defer<TResult>(delegate
			{
				QueryLanguageEx.ChainObservable<TSource> chain = null;
				return source.Select(delegate(TSource x)
				{
					QueryLanguageEx.ChainObservable<TSource> chainObservable = new QueryLanguageEx.ChainObservable<TSource>(x);
					QueryLanguageEx.ChainObservable<TSource> chain4 = chain;
					if (chain4 != null)
					{
						chain4.OnNext(chainObservable);
					}
					chain = chainObservable;
					return chainObservable;
				}).Do(delegate(IObservable<TSource> _)
				{
				}, delegate(Exception exception)
				{
					QueryLanguageEx.ChainObservable<TSource> chain2 = chain;
					if (chain2 == null)
					{
						return;
					}
					chain2.OnError(exception);
				}, delegate
				{
					QueryLanguageEx.ChainObservable<TSource> chain3 = chain;
					if (chain3 == null)
					{
						return;
					}
					chain3.OnCompleted();
				}).ObserveOn(scheduler)
					.Select(selector);
			});
		}

		public virtual ListObservable<TSource> ToListObservable<TSource>(IObservable<TSource> source)
		{
			return new ListObservable<TSource>(source);
		}

		private static IObservable<TResult> Combine<TLeft, TRight, TResult>(IObservable<TLeft> leftSource, IObservable<TRight> rightSource, Func<IObserver<TResult>, IDisposable, IDisposable, IObserver<Either<Notification<TLeft>, Notification<TRight>>>> combinerSelector)
		{
			return new QueryLanguageEx.CombineObservable<TLeft, TRight, TResult>(leftSource, rightSource, combinerSelector);
		}

		private sealed class CreateWithEnumerableObservable<TResult> : ObservableBase<TResult>
		{
			public CreateWithEnumerableObservable(Func<IObserver<TResult>, IEnumerable<IObservable<object>>> iteratorMethod)
			{
				this._iteratorMethod = iteratorMethod;
			}

			protected override IDisposable SubscribeCore(IObserver<TResult> observer)
			{
				return this._iteratorMethod(observer).Concat<object>().Subscribe(new QueryLanguageEx.TerminalOnlyObserver<TResult>(observer));
			}

			private readonly Func<IObserver<TResult>, IEnumerable<IObservable<object>>> _iteratorMethod;
		}

		private sealed class TerminalOnlyObserver<TResult> : IObserver<object>
		{
			public TerminalOnlyObserver(IObserver<TResult> observer)
			{
				this._observer = observer;
			}

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

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

			public void OnNext(object value)
			{
			}

			private readonly IObserver<TResult> _observer;
		}

		private sealed class CreateWithOnlyEnumerableObservable<TResult> : ObservableBase<TResult>
		{
			public CreateWithOnlyEnumerableObservable(Func<IEnumerable<IObservable<object>>> iteratorMethod)
			{
				this._iteratorMethod = iteratorMethod;
			}

			protected override IDisposable SubscribeCore(IObserver<TResult> observer)
			{
				return this._iteratorMethod().Concat<object>().Subscribe(new QueryLanguageEx.TerminalOnlyObserver<TResult>(observer));
			}

			private readonly Func<IEnumerable<IObservable<object>>> _iteratorMethod;
		}

		private sealed class ExpandObservable<TSource> : ObservableBase<TSource>
		{
			public ExpandObservable(IObservable<TSource> source, Func<TSource, IObservable<TSource>> selector, IScheduler scheduler)
			{
				this._source = source;
				this._selector = selector;
				this._scheduler = scheduler;
			}

			protected override IDisposable SubscribeCore(IObserver<TSource> observer)
			{
				QueryLanguageEx.ExpandObservable<TSource>.<>c__DisplayClass4_0 CS$<>8__locals1 = new QueryLanguageEx.ExpandObservable<TSource>.<>c__DisplayClass4_0();
				CS$<>8__locals1.<>4__this = this;
				CS$<>8__locals1.observer = observer;
				CS$<>8__locals1.outGate = new object();
				CS$<>8__locals1.q = new Queue<IObservable<TSource>>();
				CS$<>8__locals1.m = new SerialDisposable();
				CS$<>8__locals1.d = new CompositeDisposable { CS$<>8__locals1.m };
				CS$<>8__locals1.activeCount = 0;
				CS$<>8__locals1.isAcquired = false;
				Queue<IObservable<TSource>> q = CS$<>8__locals1.q;
				lock (q)
				{
					CS$<>8__locals1.q.Enqueue(this._source);
					int activeCount = CS$<>8__locals1.activeCount;
					CS$<>8__locals1.activeCount = activeCount + 1;
				}
				CS$<>8__locals1.method_0();
				return CS$<>8__locals1.d;
			}

			private readonly IObservable<TSource> _source;

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

			private readonly IScheduler _scheduler;
		}

		private sealed class ForkJoinObservable<TSource> : ObservableBase<TSource[]>
		{
			public ForkJoinObservable(IEnumerable<IObservable<TSource>> sources)
			{
				this._sources = sources;
			}

			protected override IDisposable SubscribeCore(IObserver<TSource[]> observer)
			{
				IObservable<TSource>[] array = this._sources.ToArray<IObservable<TSource>>();
				int num = array.Length;
				if (num == 0)
				{
					observer.OnCompleted();
					return Disposable.Empty;
				}
				CompositeDisposable group = new CompositeDisposable(array.Length);
				object gate = new object();
				bool finished = false;
				bool[] hasResults = new bool[num];
				bool[] hasCompleted = new bool[num];
				List<TSource> results = new List<TSource>(num);
				object gate5 = gate;
				lock (gate5)
				{
					Action<Exception> <>9__1;
					for (int i = 0; i < num; i++)
					{
						int currentIndex = i;
						IObservable<TSource> observable = array[i];
						results.Add(default(TSource));
						CompositeDisposable group2 = group;
						IObservable<TSource> observable2 = observable;
						Action<TSource> action = delegate(TSource value)
						{
							object gate2 = gate;
							lock (gate2)
							{
								if (!finished)
								{
									hasResults[currentIndex] = true;
									results[currentIndex] = value;
								}
							}
						};
						Action<Exception> action2;
						if ((action2 = <>9__1) == null)
						{
							action2 = (<>9__1 = delegate(Exception error)
							{
								object gate3 = gate;
								lock (gate3)
								{
									finished = true;
									observer.OnError(error);
									group.Dispose();
								}
							});
						}
						group2.Add(observable2.Subscribe(action, action2, delegate
						{
							object gate4 = gate;
							lock (gate4)
							{
								if (!finished)
								{
									if (!hasResults[currentIndex])
									{
										observer.OnCompleted();
									}
									else
									{
										hasCompleted[currentIndex] = true;
										bool[] hasCompleted2 = hasCompleted;
										for (int j = 0; j < hasCompleted2.Length; j++)
										{
											if (!hasCompleted2[j])
											{
												return;
											}
										}
										finished = true;
										observer.OnNext(results.ToArray());
										observer.OnCompleted();
									}
								}
							}
						}));
					}
				}
				return group;
			}

			private readonly IEnumerable<IObservable<TSource>> _sources;
		}

		private class ChainObservable<T> : ISubject<IObservable<T>, T>, IObserver<IObservable<T>>, IObservable<T>
		{
			public ChainObservable(T head)
			{
				this._head = head;
			}

			public IDisposable Subscribe(IObserver<T> observer)
			{
				CompositeDisposable compositeDisposable = new CompositeDisposable();
				compositeDisposable.Add(CurrentThreadScheduler.Instance.ScheduleAction(new ValueTuple<IObserver<T>, CompositeDisposable, QueryLanguageEx.ChainObservable<T>>(observer, compositeDisposable, this), delegate([TupleElementNames(new string[] { "observer", "g", "this" })] ValueTuple<IObserver<T>, CompositeDisposable, QueryLanguageEx.ChainObservable<T>> state)
				{
					state.Item1.OnNext(state.Item3._head);
					state.Item2.Add(state.Item3._tail.Merge<T>().Subscribe(state.Item1));
				}));
				return compositeDisposable;
			}

			public void OnCompleted()
			{
				this.OnNext(Observable.Empty<T>());
			}

			public void OnError(Exception error)
			{
				this.OnNext(Observable.Throw<T>(error));
			}

			public void OnNext(IObservable<T> value)
			{
				this._tail.OnNext(value);
				this._tail.OnCompleted();
			}

			private readonly T _head;

			private readonly AsyncSubject<IObservable<T>> _tail = new AsyncSubject<IObservable<T>>();
		}

		private sealed class CombineObservable<TLeft, TRight, TResult> : ObservableBase<TResult>
		{
			public CombineObservable(IObservable<TLeft> leftSource, IObservable<TRight> rightSource, Func<IObserver<TResult>, IDisposable, IDisposable, IObserver<Either<Notification<TLeft>, Notification<TRight>>>> combinerSelector)
			{
				this._leftSource = leftSource;
				this._rightSource = rightSource;
				this._combinerSelector = combinerSelector;
			}

			protected override IDisposable SubscribeCore(IObserver<TResult> observer)
			{
				SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable();
				SingleAssignmentDisposable singleAssignmentDisposable2 = new SingleAssignmentDisposable();
				IObserver<Either<Notification<TLeft>, Notification<TRight>>> observer2 = this._combinerSelector(observer, singleAssignmentDisposable, singleAssignmentDisposable2);
				object obj = new object();
				singleAssignmentDisposable.Disposable = (from x in this._leftSource.Materialize<TLeft>()
					select Either<Notification<TLeft>, Notification<TRight>>.CreateLeft(x)).Synchronize(obj).Subscribe(observer2);
				singleAssignmentDisposable2.Disposable = (from x in this._rightSource.Materialize<TRight>()
					select Either<Notification<TLeft>, Notification<TRight>>.CreateRight(x)).Synchronize(obj).Subscribe(observer2);
				return StableCompositeDisposable.Create(singleAssignmentDisposable, singleAssignmentDisposable2);
			}

			private readonly IObservable<TLeft> _leftSource;

			private readonly IObservable<TRight> _rightSource;

			private readonly Func<IObserver<TResult>, IDisposable, IDisposable, IObserver<Either<Notification<TLeft>, Notification<TRight>>>> _combinerSelector;
		}
	}
}
