using System;
using System.Reactive.Concurrency;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class TimeInterval<TSource> : Producer<TimeInterval<TSource>, TimeInterval<TSource>._>
	{
		public TimeInterval(IObservable<TSource> source, IScheduler scheduler)
		{
			this._source = source;
			this._scheduler = scheduler;
		}

		protected override TimeInterval<TSource>._ CreateSink(IObserver<TimeInterval<TSource>> observer)
		{
			return new TimeInterval<TSource>._(observer);
		}

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

		private readonly IObservable<TSource> _source;

		private readonly IScheduler _scheduler;

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

			public void Run(TimeInterval<TSource> parent)
			{
				this._watch = parent._scheduler.StartStopwatch();
				this._last = TimeSpan.Zero;
				base.SetUpstream(parent._source.Subscribe(this));
			}

			public override void OnNext(TSource value)
			{
				TimeSpan elapsed = this._watch.Elapsed;
				TimeSpan timeSpan = elapsed.Subtract(this._last);
				this._last = elapsed;
				base.ForwardOnNext(new TimeInterval<TSource>(value, timeSpan));
			}

			private IStopwatch _watch;

			private TimeSpan _last;
		}
	}
}
