using System;
using System.Collections.Generic;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Subjects;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class GroupByUntil<TSource, TKey, TElement, T> : Producer<GInterface4<TKey, TElement>, GroupByUntil<TSource, TKey, TElement, T>._>
	{
		public GroupByUntil(IObservable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, Func<GInterface4<TKey, TElement>, IObservable<T>> durationSelector, int? capacity, IEqualityComparer<TKey> comparer)
		{
			this._source = source;
			this._keySelector = keySelector;
			this._elementSelector = elementSelector;
			this._durationSelector = durationSelector;
			this._capacity = capacity;
			this._comparer = comparer;
		}

		protected override GroupByUntil<TSource, TKey, TElement, T>._ CreateSink(IObserver<GInterface4<TKey, TElement>> observer)
		{
			return new GroupByUntil<TSource, TKey, TElement, T>._(this, observer);
		}

		protected override void Run(GroupByUntil<TSource, TKey, TElement, T>._ sink)
		{
			sink.Run(this._source);
		}

		private readonly IObservable<TSource> _source;

		private readonly Func<TSource, TKey> _keySelector;

		private readonly Func<TSource, TElement> _elementSelector;

		private readonly Func<GInterface4<TKey, TElement>, IObservable<T>> _durationSelector;

		private readonly int? _capacity;

		private readonly IEqualityComparer<TKey> _comparer;

		internal sealed class _ : Sink<TSource, GInterface4<TKey, TElement>>
		{
			public _(GroupByUntil<TSource, TKey, TElement, T> parent, IObserver<GInterface4<TKey, TElement>> observer)
				: base(observer)
			{
				this._refCountDisposable = new RefCountDisposable(this._groupDisposable);
				this._map = new Map<TKey, ISubject<TElement>>(parent._capacity, parent._comparer);
				this._keySelector = parent._keySelector;
				this._elementSelector = parent._elementSelector;
				this._durationSelector = parent._durationSelector;
			}

			public override void Run(IObservable<TSource> source)
			{
				this._groupDisposable.Add(source.SubscribeSafe(this));
				base.SetUpstream(this._refCountDisposable);
			}

			private ISubject<TElement> NewSubject()
			{
				Subject<TElement> subject = new Subject<TElement>();
				return Subject.Create<TElement>(new AsyncLockObserver<TElement>(subject, new AsyncLock()), subject);
			}

			public override void OnNext(TSource value)
			{
				TKey tkey;
				try
				{
					tkey = this._keySelector(value);
				}
				catch (Exception ex)
				{
					this.Error(ex);
					return;
				}
				bool flag = false;
				ISubject<TElement> subject = null;
				try
				{
					if (tkey == null)
					{
						object obj = this._nullGate;
						lock (obj)
						{
							if (this._null == null)
							{
								this._null = this.NewSubject();
								flag = true;
							}
							subject = this._null;
							goto IL_0084;
						}
					}
					subject = this._map.GetOrAdd(tkey, new Func<ISubject<TElement>>(this.NewSubject), out flag);
					IL_0084:;
				}
				catch (Exception ex2)
				{
					this.Error(ex2);
					return;
				}
				if (flag)
				{
					GroupedObservable<TKey, TElement> groupedObservable = new GroupedObservable<TKey, TElement>(tkey, subject, this._refCountDisposable);
					IObservable<T> observable = null;
					GroupedObservable<TKey, TElement> groupedObservable2 = new GroupedObservable<TKey, TElement>(tkey, subject);
					try
					{
						observable = this._durationSelector(groupedObservable2);
					}
					catch (Exception ex3)
					{
						this.Error(ex3);
						return;
					}
					object obj = this._gate;
					lock (obj)
					{
						base.ForwardOnNext(groupedObservable);
					}
					GroupByUntil<TSource, TKey, TElement, T>._.DurationObserver durationObserver = new GroupByUntil<TSource, TKey, TElement, T>._.DurationObserver(this, tkey, subject);
					this._groupDisposable.Add(durationObserver);
					durationObserver.SetResource(observable.SubscribeSafe(durationObserver));
				}
				TElement telement;
				try
				{
					telement = this._elementSelector(value);
				}
				catch (Exception ex4)
				{
					this.Error(ex4);
					return;
				}
				subject.OnNext(telement);
			}

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

			public override void OnCompleted()
			{
				ISubject<TElement> subject = null;
				object obj = this._nullGate;
				lock (obj)
				{
					subject = this._null;
				}
				if (subject != null)
				{
					subject.OnCompleted();
				}
				foreach (ISubject<TElement> subject2 in this._map.Values)
				{
					subject2.OnCompleted();
				}
				obj = this._gate;
				lock (obj)
				{
					base.ForwardOnCompleted();
				}
			}

			private void Error(Exception exception)
			{
				ISubject<TElement> subject = null;
				object obj = this._nullGate;
				lock (obj)
				{
					subject = this._null;
				}
				if (subject != null)
				{
					subject.OnError(exception);
				}
				foreach (ISubject<TElement> subject2 in this._map.Values)
				{
					subject2.OnError(exception);
				}
				obj = this._gate;
				lock (obj)
				{
					base.ForwardOnError(exception);
				}
			}

			private readonly object _gate = new object();

			private readonly object _nullGate = new object();

			private readonly CompositeDisposable _groupDisposable = new CompositeDisposable();

			private readonly RefCountDisposable _refCountDisposable;

			private readonly Map<TKey, ISubject<TElement>> _map;

			private readonly Func<TSource, TKey> _keySelector;

			private readonly Func<TSource, TElement> _elementSelector;

			private readonly Func<GInterface4<TKey, TElement>, IObservable<T>> _durationSelector;

			private ISubject<TElement> _null;

			private sealed class DurationObserver : SafeObserver<T>
			{
				public DurationObserver(GroupByUntil<TSource, TKey, TElement, T>._ parent, TKey key, ISubject<TElement> writer)
				{
					this._parent = parent;
					this._key = key;
					this._writer = writer;
				}

				public override void OnNext(T value)
				{
					this.OnCompleted();
				}

				public override void OnError(Exception error)
				{
					this._parent.Error(error);
					base.Dispose();
				}

				public override void OnCompleted()
				{
					if (this._key == null)
					{
						ISubject<TElement> subject = null;
						object nullGate = this._parent._nullGate;
						lock (nullGate)
						{
							subject = this._parent._null;
							this._parent._null = null;
						}
						subject.OnCompleted();
					}
					else if (this._parent._map.Remove(this._key))
					{
						this._writer.OnCompleted();
					}
					this._parent._groupDisposable.Remove(this);
				}

				private readonly GroupByUntil<TSource, TKey, TElement, T>._ _parent;

				private readonly TKey _key;

				private readonly ISubject<TElement> _writer;
			}
		}
	}
}
