using System;
using System.Reactive.Concurrency;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Throw<TResult> : Producer<TResult, Throw<TResult>._>
	{
		public Throw(Exception exception, IScheduler scheduler)
		{
			this._exception = exception;
			this._scheduler = scheduler;
		}

		protected override Throw<TResult>._ CreateSink(IObserver<TResult> observer)
		{
			return new Throw<TResult>._(this._exception, observer);
		}

		protected override void Run(Throw<TResult>._ sink)
		{
			sink.Run(this._scheduler);
		}

		private readonly Exception _exception;

		private readonly IScheduler _scheduler;

		internal sealed class _ : IdentitySink<TResult>
		{
			public _(Exception exception, IObserver<TResult> observer)
				: base(observer)
			{
				this._exception = exception;
			}

			public void Run(IScheduler scheduler)
			{
				base.SetUpstream(scheduler.ScheduleAction(this, delegate(Throw<TResult>._ @this)
				{
					@this.ForwardOnError(@this._exception);
				}));
			}

			private readonly Exception _exception;
		}
	}
}
