using System;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;

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

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

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

		private readonly IObservable<TSource> _source;

		private readonly TimeSpan _interval;

		private readonly IScheduler _scheduler;

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

			public void Run(Sample<TSource> parent)
			{
				Disposable.SetSingle(ref this._sourceDisposable, parent._source.SubscribeSafe(this));
				base.SetUpstream(parent._scheduler.SchedulePeriodic(parent._interval, new Action(this.Tick)));
			}

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

			private void Tick()
			{
				object gate = this._gate;
				lock (gate)
				{
					if (this._hasValue)
					{
						this._hasValue = false;
						base.ForwardOnNext(this._value);
					}
					if (this._atEnd)
					{
						base.ForwardOnCompleted();
					}
				}
			}

			public override void OnNext(TSource value)
			{
				object gate = this._gate;
				lock (gate)
				{
					this._hasValue = true;
					this._value = value;
				}
			}

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

			public override void OnCompleted()
			{
				object gate = this._gate;
				lock (gate)
				{
					this._atEnd = true;
					Disposable.TryDispose(ref this._sourceDisposable);
				}
			}

			private readonly object _gate = new object();

			private IDisposable _sourceDisposable;

			private bool _hasValue;

			private TSource _value;

			private bool _atEnd;
		}
	}
}
