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

namespace System.Reactive.Linq.ObservableImpl
{
	internal static class Take<TSource>
	{
		internal sealed class Count : Producer<TSource, Take<TSource>.Count._>
		{
			public Count(IObservable<TSource> source, int count)
			{
				this._source = source;
				this._count = count;
			}

			public IObservable<TSource> Combine(int count)
			{
				if (this._count <= count)
				{
					return this;
				}
				return new Take<TSource>.Count(this._source, count);
			}

			protected override Take<TSource>.Count._ CreateSink(IObserver<TSource> observer)
			{
				return new Take<TSource>.Count._(this._count, observer);
			}

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

			private readonly IObservable<TSource> _source;

			private readonly int _count;

			internal sealed class _ : IdentitySink<TSource>
			{
				public _(int count, IObserver<TSource> observer)
					: base(observer)
				{
					this._remaining = count;
				}

				public override void OnNext(TSource value)
				{
					if (this._remaining > 0)
					{
						this._remaining--;
						base.ForwardOnNext(value);
						if (this._remaining == 0)
						{
							base.ForwardOnCompleted();
						}
					}
				}

				private int _remaining;
			}
		}

		internal sealed class Time : Producer<TSource, Take<TSource>.Time._>
		{
			public Time(IObservable<TSource> source, TimeSpan duration, IScheduler scheduler)
			{
				this._source = source;
				this._duration = duration;
				this._scheduler = scheduler;
			}

			public IObservable<TSource> Combine(TimeSpan duration)
			{
				if (this._duration <= duration)
				{
					return this;
				}
				return new Take<TSource>.Time(this._source, duration, this._scheduler);
			}

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

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

			private readonly IObservable<TSource> _source;

			private readonly TimeSpan _duration;

			internal readonly IScheduler _scheduler;

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

				public void Run(Take<TSource>.Time parent)
				{
					this._gate = new object();
					Disposable.SetSingle(ref this._task, parent._scheduler.ScheduleAction(this, parent._duration, delegate(Take<TSource>.Time._ 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()
				{
					object gate = this._gate;
					lock (gate)
					{
						base.ForwardOnCompleted();
					}
				}

				public override void OnNext(TSource value)
				{
					object gate = this._gate;
					lock (gate)
					{
						base.ForwardOnNext(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)
					{
						base.ForwardOnCompleted();
					}
				}

				private object _gate;

				private IDisposable _task;
			}
		}
	}
}
