using System;
using System.Reactive.Concurrency;

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

		protected override Timestamp<TSource>._ CreateSink(IObserver<Timestamped<TSource>> observer)
		{
			return new Timestamp<TSource>._(this._scheduler, observer);
		}

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

		private readonly IObservable<TSource> _source;

		private readonly IScheduler _scheduler;

		internal sealed class _ : Sink<TSource, Timestamped<TSource>>
		{
			public _(IScheduler scheduler, IObserver<Timestamped<TSource>> observer)
				: base(observer)
			{
				this._scheduler = scheduler;
			}

			public override void OnNext(TSource value)
			{
				base.ForwardOnNext(new Timestamped<TSource>(value, this._scheduler.Now));
			}

			private readonly IScheduler _scheduler;
		}
	}
}
