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

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Zip<TSource> : Producer<IList<TSource>, Zip<TSource>._>
	{
		public Zip(IEnumerable<IObservable<TSource>> sources)
		{
			this._sources = sources;
		}

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

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

		private readonly IEnumerable<IObservable<TSource>> _sources;

		internal sealed class _ : IdentitySink<IList<TSource>>
		{
			public _(IObserver<IList<TSource>> observer)
				: base(observer)
			{
				this._gate = new object();
			}

			public void Run(IEnumerable<IObservable<TSource>> sources)
			{
				IObservable<TSource>[] array = sources.ToArray<IObservable<TSource>>();
				int num = array.Length;
				this._queues = new Queue<TSource>[num];
				for (int i = 0; i < num; i++)
				{
					this._queues[i] = new Queue<TSource>();
				}
				this._isDone = new bool[num];
				IDisposable[] array2 = new IDisposable[num];
				if (Interlocked.CompareExchange<IDisposable[]>(ref this._subscriptions, array2, null) == null)
				{
					for (int j = 0; j < num; j++)
					{
						Zip<TSource>._.SourceObserver sourceObserver = new Zip<TSource>._.SourceObserver(this, j);
						Disposable.SetSingle(ref array2[j], array[j].SubscribeSafe(sourceObserver));
					}
				}
			}

			protected override void Dispose(bool disposing)
			{
				if (disposing)
				{
					IDisposable[] array = Interlocked.Exchange<IDisposable[]>(ref this._subscriptions, Array.Empty<IDisposable>());
					if (array != null && array != Array.Empty<IDisposable>())
					{
						for (int i = 0; i < array.Length; i++)
						{
							Disposable.TryDispose(ref array[i]);
						}
						object gate = this._gate;
						lock (gate)
						{
							Queue<TSource>[] queues = this._queues;
							for (int j = 0; j < queues.Length; j++)
							{
								queues[j].Clear();
							}
						}
					}
				}
				base.Dispose(disposing);
			}

			private void OnNext(int index, TSource value)
			{
				object gate = this._gate;
				lock (gate)
				{
					this._queues[index].Enqueue(value);
					if (this._queues.All((Queue<TSource> q) => q.Count > 0))
					{
						int num = this._queues.Length;
						List<TSource> list = new List<TSource>(num);
						for (int i = 0; i < num; i++)
						{
							list.Add(this._queues[i].Dequeue());
						}
						base.ForwardOnNext(list);
					}
					else if (this._isDone.AllExcept(index))
					{
						base.ForwardOnCompleted();
					}
				}
			}

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

			private void OnCompleted(int index)
			{
				object gate = this._gate;
				lock (gate)
				{
					this._isDone[index] = true;
					if (this._isDone.All())
					{
						base.ForwardOnCompleted();
					}
					else
					{
						IDisposable[] array = Volatile.Read<IDisposable[]>(ref this._subscriptions);
						if (array != null && array != Array.Empty<IDisposable>())
						{
							Disposable.TryDispose(ref array[index]);
						}
					}
				}
			}

			private readonly object _gate;

			private Queue<TSource>[] _queues;

			private bool[] _isDone;

			private IDisposable[] _subscriptions;

			private sealed class SourceObserver : IObserver<TSource>
			{
				public SourceObserver(Zip<TSource>._ parent, int index)
				{
					this._parent = parent;
					this._index = index;
				}

				public void OnNext(TSource value)
				{
					this._parent.OnNext(this._index, value);
				}

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

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

				private readonly Zip<TSource>._ _parent;

				private readonly int _index;
			}
		}
	}
}
