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

namespace System.Reactive.Joins
{
	internal sealed class JoinObserver<T> : ObserverBase<Notification<T>>, IJoinObserver, IDisposable
	{
		public Queue<Notification<T>> Queue { get; }

		public JoinObserver(IObservable<T> source, Action<Exception> onError)
		{
			this._source = source;
			this._onError = onError;
			this.Queue = new Queue<Notification<T>>();
			this._activePlans = new List<ActivePlan>();
		}

		public void AddActivePlan(ActivePlan activePlan)
		{
			this._activePlans.Add(activePlan);
		}

		public void Subscribe(object gate)
		{
			this._gate = gate;
			Disposable.SetSingle(ref this._subscription, this._source.Materialize<T>().SubscribeSafe(this));
		}

		public void Dequeue()
		{
			this.Queue.Dequeue();
		}

		protected override void OnNextCore(Notification<T> notification)
		{
			object gate = this._gate;
			lock (gate)
			{
				if (!this._isDisposed)
				{
					if (notification.Kind == NotificationKind.OnError)
					{
						this._onError(notification.Exception);
					}
					else
					{
						this.Queue.Enqueue(notification);
						ActivePlan[] array = this._activePlans.ToArray();
						for (int i = 0; i < array.Length; i++)
						{
							array[i].Match();
						}
					}
				}
			}
		}

		protected override void OnErrorCore(Exception exception)
		{
		}

		protected override void OnCompletedCore()
		{
		}

		internal void RemoveActivePlan(ActivePlan activePlan)
		{
			this._activePlans.Remove(activePlan);
			if (this._activePlans.Count == 0)
			{
				base.Dispose();
			}
		}

		protected override void Dispose(bool disposing)
		{
			base.Dispose(disposing);
			if (!this._isDisposed)
			{
				if (disposing)
				{
					Disposable.TryDispose(ref this._subscription);
				}
				this._isDisposed = true;
			}
		}

		private object _gate;

		private readonly IObservable<T> _source;

		private readonly Action<Exception> _onError;

		private readonly List<ActivePlan> _activePlans;

		private IDisposable _subscription;

		private bool _isDisposed;
	}
}
