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

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

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

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

			private readonly IObservable<TSource> _source;

			private readonly int _count;

			internal sealed class _ : IdentitySink<TSource>
			{
				public _(int count, IObserver<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)
					{
						base.ForwardOnNext(this._queue.Dequeue());
					}
				}

				private readonly int _count;

				private readonly Queue<TSource> _queue;
			}
		}

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

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

			protected override void Run(SkipLast<TSource>.Time._ sink)
			{
				sink.Run(this);
			}

			private readonly IObservable<TSource> _source;

			private readonly TimeSpan _duration;

			private readonly IScheduler _scheduler;

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

				public void Run(SkipLast<TSource>.Time parent)
				{
					this._watch = parent._scheduler.StartStopwatch();
					base.SetUpstream(parent._source.SubscribeSafe(this));
				}

				public override void OnNext(TSource value)
				{
					TimeSpan elapsed = this._watch.Elapsed;
					this._queue.Enqueue(new TimeInterval<TSource>(value, elapsed));
					while (this._queue.Count > 0 && elapsed - this._queue.Peek().Interval >= this._duration)
					{
						base.ForwardOnNext(this._queue.Dequeue().Value);
					}
				}

				public override void OnCompleted()
				{
					TimeSpan elapsed = this._watch.Elapsed;
					while (this._queue.Count > 0 && elapsed - this._queue.Peek().Interval >= this._duration)
					{
						base.ForwardOnNext(this._queue.Dequeue().Value);
					}
					base.ForwardOnCompleted();
				}

				private readonly TimeSpan _duration;

				private readonly Queue<TimeInterval<TSource>> _queue;

				private IStopwatch _watch;
			}
		}
	}
}
