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

namespace System.Reactive.Linq.ObservableImpl
{
	internal static class TakeLastBuffer<TSource>
	{
		internal sealed class Count : Producer<IList<TSource>, TakeLastBuffer<TSource>.Count._>
		{
			public Count(IObservable<TSource> source, int count)
			{
				this._source = source;
				this._count = count;
			}

			protected override TakeLastBuffer<TSource>.Count._ CreateSink(IObserver<IList<TSource>> observer)
			{
				return new TakeLastBuffer<TSource>.Count._(this._count, observer);
			}

			protected override void Run(TakeLastBuffer<TSource>.Count._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly int _count;

			internal sealed class _ : Sink<TSource, IList<TSource>>
			{
				public _(int count, IObserver<IList<TSource>> observer)
					: base(observer)
				{
					this._count = count;
					this._queue = new Queue<TSource>();
				}

				public override void OnNext(TSource value)
				{
					this._queue.Enqueue(value);
					if (this._queue.Count > this._count)
					{
						this._queue.Dequeue();
					}
				}

				public override void OnCompleted()
				{
					List<TSource> list = new List<TSource>(this._queue.Count);
					while (this._queue.Count > 0)
					{
						list.Add(this._queue.Dequeue());
					}
					base.ForwardOnNext(list);
					base.ForwardOnCompleted();
				}

				private readonly int _count;

				private readonly Queue<TSource> _queue;
			}
		}

		internal sealed class Time : Producer<IList<TSource>, TakeLastBuffer<TSource>.Time._>
		{
			public Time(IObservable<TSource> source, TimeSpan duration, IScheduler scheduler)
			{
				this._source = source;
				this._duration = duration;
				this._scheduler = scheduler;
			}

			protected override TakeLastBuffer<TSource>.Time._ CreateSink(IObserver<IList<TSource>> observer)
			{
				return new TakeLastBuffer<TSource>.Time._(this._duration, observer);
			}

			protected override void Run(TakeLastBuffer<TSource>.Time._ sink)
			{
				sink.Run(this._source, this._scheduler);
			}

			private readonly IObservable<TSource> _source;

			private readonly TimeSpan _duration;

			private readonly IScheduler _scheduler;

			internal sealed class _ : Sink<TSource, IList<TSource>>
			{
				public _(TimeSpan duration, IObserver<IList<TSource>> observer)
					: base(observer)
				{
					this._duration = duration;
					this._queue = new Queue<TimeInterval<TSource>>();
				}

				public void Run(IObservable<TSource> source, IScheduler scheduler)
				{
					this._watch = scheduler.StartStopwatch();
					this.Run(source);
				}

				public override void OnNext(TSource value)
				{
					TimeSpan elapsed = this._watch.Elapsed;
					this._queue.Enqueue(new TimeInterval<TSource>(value, elapsed));
					this.Trim(elapsed);
				}

				public override void OnCompleted()
				{
					TimeSpan elapsed = this._watch.Elapsed;
					this.Trim(elapsed);
					List<TSource> list = new List<TSource>(this._queue.Count);
					while (this._queue.Count > 0)
					{
						list.Add(this._queue.Dequeue().Value);
					}
					base.ForwardOnNext(list);
					base.ForwardOnCompleted();
				}

				private void Trim(TimeSpan now)
				{
					while (this._queue.Count > 0 && now - this._queue.Peek().Interval >= this._duration)
					{
						this._queue.Dequeue();
					}
				}

				private readonly TimeSpan _duration;

				private readonly Queue<TimeInterval<TSource>> _queue;

				private IStopwatch _watch;
			}
		}
	}
}
