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

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

		public IObservable<TSource> Combine(DateTimeOffset endTime)
		{
			if (this._endTime <= endTime)
			{
				return this;
			}
			return new TakeUntil<TSource>(this._source, endTime, this._scheduler);
		}

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

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

		private readonly IObservable<TSource> _source;

		private readonly DateTimeOffset _endTime;

		internal readonly IScheduler _scheduler;

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

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

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

			private void Tick()
			{
				this.OnCompleted();
			}

			public override void OnNext(TSource value)
			{
				HalfSerializer.ForwardOnNext<TSource>(this, value, ref this._wip, ref this._error);
			}

			public override void OnError(Exception error)
			{
				HalfSerializer.ForwardOnError<TSource>(this, error, ref this._wip, ref this._error);
			}

			public override void OnCompleted()
			{
				HalfSerializer.ForwardOnCompleted<TSource>(this, ref this._wip, ref this._error);
			}

			private IDisposable _timerDisposable;

			private int _wip;

			private Exception _error;
		}
	}
}
