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

namespace System.Reactive.Linq.ObservableImpl
{
	internal static class Window<TSource, TWindowClosing>
	{
		internal sealed class Selector : Producer<IObservable<TSource>, Window<TSource, TWindowClosing>.Selector._>
		{
			public Selector(IObservable<TSource> source, Func<IObservable<TWindowClosing>> windowClosingSelector)
			{
				this._source = source;
				this._windowClosingSelector = windowClosingSelector;
			}

			protected override Window<TSource, TWindowClosing>.Selector._ CreateSink(IObserver<IObservable<TSource>> observer)
			{
				return new Window<TSource, TWindowClosing>.Selector._(this, observer);
			}

			protected override void Run(Window<TSource, TWindowClosing>.Selector._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly Func<IObservable<TWindowClosing>> _windowClosingSelector;

			internal sealed class _ : Sink<TSource, IObservable<TSource>>
			{
				public _(Window<TSource, TWindowClosing>.Selector parent, IObserver<IObservable<TSource>> observer)
					: base(observer)
				{
					this._windowClosingSelector = parent._windowClosingSelector;
				}

				public override void Run(IObservable<TSource> source)
				{
					this._window = new Subject<TSource>();
					CompositeDisposable compositeDisposable = new CompositeDisposable(2) { this._m };
					this._refCountDisposable = new RefCountDisposable(compositeDisposable);
					WindowObservable<TSource> windowObservable = new WindowObservable<TSource>(this._window, this._refCountDisposable);
					base.ForwardOnNext(windowObservable);
					compositeDisposable.Add(source.SubscribeSafe(this));
					this._windowGate.Wait<Window<TSource, TWindowClosing>.Selector._>(this, delegate(Window<TSource, TWindowClosing>.Selector._ @this)
					{
						@this.CreateWindowClose();
					});
					base.SetUpstream(this._refCountDisposable);
				}

				private void CreateWindowClose()
				{
					IObservable<TWindowClosing> observable = null;
					try
					{
						observable = this._windowClosingSelector();
					}
					catch (Exception ex)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnError(ex);
						}
						return;
					}
					Window<TSource, TWindowClosing>.Selector._.WindowClosingObserver windowClosingObserver = new Window<TSource, TWindowClosing>.Selector._.WindowClosingObserver(this);
					this._m.Disposable = windowClosingObserver;
					windowClosingObserver.SetResource(observable.SubscribeSafe(windowClosingObserver));
				}

				private void CloseWindow(IDisposable closingSubscription)
				{
					closingSubscription.Dispose();
					object gate = this._gate;
					lock (gate)
					{
						this._window.OnCompleted();
						this._window = new Subject<TSource>();
						WindowObservable<TSource> windowObservable = new WindowObservable<TSource>(this._window, this._refCountDisposable);
						base.ForwardOnNext(windowObservable);
					}
					this._windowGate.Wait<Window<TSource, TWindowClosing>.Selector._>(this, delegate(Window<TSource, TWindowClosing>.Selector._ @this)
					{
						@this.CreateWindowClose();
					});
				}

				public override void OnNext(TSource value)
				{
					object gate = this._gate;
					lock (gate)
					{
						this._window.OnNext(value);
					}
				}

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

				public override void OnCompleted()
				{
					object gate = this._gate;
					lock (gate)
					{
						this._window.OnCompleted();
						base.ForwardOnCompleted();
					}
				}

				private readonly object _gate = new object();

				private readonly AsyncLock _windowGate = new AsyncLock();

				private readonly SerialDisposable _m = new SerialDisposable();

				private readonly Func<IObservable<TWindowClosing>> _windowClosingSelector;

				private ISubject<TSource> _window;

				private RefCountDisposable _refCountDisposable;

				private sealed class WindowClosingObserver : SafeObserver<TWindowClosing>
				{
					public WindowClosingObserver(Window<TSource, TWindowClosing>.Selector._ parent)
					{
						this._parent = parent;
					}

					public override void OnNext(TWindowClosing value)
					{
						this._parent.CloseWindow(this);
					}

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

					public override void OnCompleted()
					{
						this._parent.CloseWindow(this);
					}

					private readonly Window<TSource, TWindowClosing>.Selector._ _parent;
				}
			}
		}

		internal sealed class Boundaries : Producer<IObservable<TSource>, Window<TSource, TWindowClosing>.Boundaries._>
		{
			public Boundaries(IObservable<TSource> source, IObservable<TWindowClosing> windowBoundaries)
			{
				this._source = source;
				this._windowBoundaries = windowBoundaries;
			}

			protected override Window<TSource, TWindowClosing>.Boundaries._ CreateSink(IObserver<IObservable<TSource>> observer)
			{
				return new Window<TSource, TWindowClosing>.Boundaries._(observer);
			}

			protected override void Run(Window<TSource, TWindowClosing>.Boundaries._ sink)
			{
				sink.Run(this);
			}

			private readonly IObservable<TSource> _source;

			private readonly IObservable<TWindowClosing> _windowBoundaries;

			internal sealed class _ : Sink<TSource, IObservable<TSource>>
			{
				public _(IObserver<IObservable<TSource>> observer)
					: base(observer)
				{
				}

				public void Run(Window<TSource, TWindowClosing>.Boundaries parent)
				{
					this._window = new Subject<TSource>();
					CompositeDisposable compositeDisposable = new CompositeDisposable(2);
					this._refCountDisposable = new RefCountDisposable(compositeDisposable);
					WindowObservable<TSource> windowObservable = new WindowObservable<TSource>(this._window, this._refCountDisposable);
					base.ForwardOnNext(windowObservable);
					compositeDisposable.Add(parent._source.SubscribeSafe(this));
					compositeDisposable.Add(parent._windowBoundaries.SubscribeSafe(new Window<TSource, TWindowClosing>.Boundaries._.WindowClosingObserver(this)));
					base.SetUpstream(this._refCountDisposable);
				}

				public override void OnNext(TSource value)
				{
					object gate = this._gate;
					lock (gate)
					{
						this._window.OnNext(value);
					}
				}

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

				public override void OnCompleted()
				{
					object gate = this._gate;
					lock (gate)
					{
						this._window.OnCompleted();
						base.ForwardOnCompleted();
					}
				}

				private readonly object _gate = new object();

				private ISubject<TSource> _window;

				private RefCountDisposable _refCountDisposable;

				private sealed class WindowClosingObserver : IObserver<TWindowClosing>
				{
					public WindowClosingObserver(Window<TSource, TWindowClosing>.Boundaries._ parent)
					{
						this._parent = parent;
					}

					public void OnNext(TWindowClosing value)
					{
						object gate = this._parent._gate;
						lock (gate)
						{
							this._parent._window.OnCompleted();
							this._parent._window = new Subject<TSource>();
							WindowObservable<TSource> windowObservable = new WindowObservable<TSource>(this._parent._window, this._parent._refCountDisposable);
							this._parent.ForwardOnNext(windowObservable);
						}
					}

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

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

					private readonly Window<TSource, TWindowClosing>.Boundaries._ _parent;
				}
			}
		}
	}
}
