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

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

		public IObservable<TSource> Combine(DateTimeOffset startTime)
		{
			if (startTime <= this._startTime)
			{
				return this;
			}
			return new SkipUntil<TSource>(this._source, startTime, this._scheduler);
		}

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

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

		private readonly IObservable<TSource> _source;

		private readonly DateTimeOffset _startTime;

		internal readonly IScheduler _scheduler;

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

			public void Run(SkipUntil<TSource> parent)
			{
				Disposable.SetSingle(ref this._task, parent._scheduler.ScheduleAction(this, parent._startTime, delegate(SkipUntil<TSource>._ state)
				{
					state.Tick();
				}));
				this.Run(parent._source);
			}

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

			private void Tick()
			{
				this._open = true;
			}

			public override void OnNext(TSource value)
			{
				if (this._open)
				{
					base.ForwardOnNext(value);
				}
			}

			private bool _open;

			private IDisposable _task;
		}
	}
}
