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

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class RangeLongRunning : Producer<int, RangeLongRunning.RangeSink>
	{
		public RangeLongRunning(int start, int count, ISchedulerLongRunning scheduler)
		{
			this._start = start;
			this._count = count;
			this._scheduler = scheduler;
		}

		protected override RangeLongRunning.RangeSink CreateSink(IObserver<int> observer)
		{
			return new RangeLongRunning.RangeSink(this._start, this._count, observer);
		}

		protected override void Run(RangeLongRunning.RangeSink sink)
		{
			sink.Run(this._scheduler);
		}

		private readonly int _start;

		private readonly int _count;

		private readonly ISchedulerLongRunning _scheduler;

		internal sealed class RangeSink : IdentitySink<int>
		{
			public RangeSink(int start, int count, IObserver<int> observer)
				: base(observer)
			{
				this._index = start;
				this._end = start + count;
			}

			public void Run(ISchedulerLongRunning scheduler)
			{
				base.SetUpstream(scheduler.ScheduleLongRunning<RangeLongRunning.RangeSink>(this, delegate(RangeLongRunning.RangeSink @this, ICancelable cancel)
				{
					@this.Loop(cancel);
				}));
			}

			private void Loop(ICancelable cancel)
			{
				int index = this._index;
				int end = this._end;
				while (!cancel.IsDisposed)
				{
					if (index == end)
					{
						break;
					}
					base.ForwardOnNext(index++);
				}
				if (!cancel.IsDisposed)
				{
					base.ForwardOnCompleted();
				}
			}

			private readonly int _end;

			private readonly int _index;
		}
	}
}
