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

namespace System.Reactive.Linq.ObservableImpl
{
	internal static class Buffer<TSource, TBufferClosing>
	{
		internal sealed class Selector : Producer<IList<TSource>, Buffer<TSource, TBufferClosing>.Selector._>
		{
			public Selector(IObservable<TSource> source, Func<IObservable<TBufferClosing>> bufferClosingSelector)
			{
				this._source = source;
				this._bufferClosingSelector = bufferClosingSelector;
			}

			protected override Buffer<TSource, TBufferClosing>.Selector._ CreateSink(IObserver<IList<TSource>> observer)
			{
				return new Buffer<TSource, TBufferClosing>.Selector._(this, observer);
			}

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

			private readonly IObservable<TSource> _source;

			private readonly Func<IObservable<TBufferClosing>> _bufferClosingSelector;

			internal sealed class _ : Sink<TSource, IList<TSource>>
			{
				public _(Buffer<TSource, TBufferClosing>.Selector parent, IObserver<IList<TSource>> observer)
					: base(observer)
				{
					this._bufferClosingSelector = parent._bufferClosingSelector;
				}

				public override void Run(IObservable<TSource> source)
				{
					this._buffer = new List<TSource>();
					base.Run(source);
					this._bufferGate.Wait<Buffer<TSource, TBufferClosing>.Selector._>(this, delegate(Buffer<TSource, TBufferClosing>.Selector._ @this)
					{
						@this.CreateBufferClose();
					});
				}

				protected override void Dispose(bool disposing)
				{
					if (disposing)
					{
						Disposable.TryDispose(ref this._bufferClosingSerialDisposable);
					}
					base.Dispose(disposing);
				}

				private void CreateBufferClose()
				{
					IObservable<TBufferClosing> observable = null;
					try
					{
						observable = this._bufferClosingSelector();
					}
					catch (Exception ex)
					{
						object gate = this._gate;
						lock (gate)
						{
							base.ForwardOnError(ex);
						}
						return;
					}
					Buffer<TSource, TBufferClosing>.Selector._.BufferClosingObserver bufferClosingObserver = new Buffer<TSource, TBufferClosing>.Selector._.BufferClosingObserver(this);
					Disposable.TrySetSerial(ref this._bufferClosingSerialDisposable, bufferClosingObserver);
					bufferClosingObserver.SetResource(observable.SubscribeSafe(bufferClosingObserver));
				}

				private void CloseBuffer(IDisposable closingSubscription)
				{
					closingSubscription.Dispose();
					object gate = this._gate;
					lock (gate)
					{
						IList<TSource> buffer = this._buffer;
						this._buffer = new List<TSource>();
						base.ForwardOnNext(buffer);
					}
					this._bufferGate.Wait<Buffer<TSource, TBufferClosing>.Selector._>(this, delegate(Buffer<TSource, TBufferClosing>.Selector._ @this)
					{
						@this.CreateBufferClose();
					});
				}

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

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

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

				private readonly object _gate = new object();

				private readonly AsyncLock _bufferGate = new AsyncLock();

				private IDisposable _bufferClosingSerialDisposable;

				private readonly Func<IObservable<TBufferClosing>> _bufferClosingSelector;

				private IList<TSource> _buffer;

				private sealed class BufferClosingObserver : SafeObserver<TBufferClosing>
				{
					public BufferClosingObserver(Buffer<TSource, TBufferClosing>.Selector._ parent)
					{
						this._parent = parent;
					}

					public override void OnNext(TBufferClosing value)
					{
						this._parent.CloseBuffer(this);
					}

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

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

					private readonly Buffer<TSource, TBufferClosing>.Selector._ _parent;
				}
			}
		}

		internal sealed class Boundaries : Producer<IList<TSource>, Buffer<TSource, TBufferClosing>.Boundaries._>
		{
			public Boundaries(IObservable<TSource> source, IObservable<TBufferClosing> bufferBoundaries)
			{
				this._source = source;
				this._bufferBoundaries = bufferBoundaries;
			}

			protected override Buffer<TSource, TBufferClosing>.Boundaries._ CreateSink(IObserver<IList<TSource>> observer)
			{
				return new Buffer<TSource, TBufferClosing>.Boundaries._(observer);
			}

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

			private readonly IObservable<TSource> _source;

			private readonly IObservable<TBufferClosing> _bufferBoundaries;

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

				public void Run(Buffer<TSource, TBufferClosing>.Boundaries parent)
				{
					this._buffer = new List<TSource>();
					this.Run(parent._source);
					Disposable.SetSingle(ref this._boundariesDisposable, parent._bufferBoundaries.SubscribeSafe(new Buffer<TSource, TBufferClosing>.Boundaries._.BufferClosingObserver(this)));
				}

				protected override void Dispose(bool disposing)
				{
					if (disposing)
					{
						Disposable.TryDispose(ref this._boundariesDisposable);
					}
					base.Dispose(disposing);
				}

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

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

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

				private readonly object _gate = new object();

				private IList<TSource> _buffer;

				private IDisposable _boundariesDisposable;

				private sealed class BufferClosingObserver : IObserver<TBufferClosing>
				{
					public BufferClosingObserver(Buffer<TSource, TBufferClosing>.Boundaries._ parent)
					{
						this._parent = parent;
					}

					public void OnNext(TBufferClosing value)
					{
						object gate = this._parent._gate;
						lock (gate)
						{
							IList<TSource> buffer = this._parent._buffer;
							this._parent._buffer = new List<TSource>();
							this._parent.ForwardOnNext(buffer);
						}
					}

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

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

					private readonly Buffer<TSource, TBufferClosing>.Boundaries._ _parent;
				}
			}
		}
	}
}
