using System;

namespace System.Reactive
{
	internal abstract class Sink<TSource, TTarget> : Sink<TTarget>, IObserver<TSource>
	{
		protected Sink(IObserver<TTarget> observer)
			: base(observer)
		{
		}

		public virtual void Run(IObservable<TSource> source)
		{
			base.SetUpstream(source.SubscribeSafe(this));
		}

		public abstract void OnNext(TSource value);

		public virtual void OnError(Exception error)
		{
			base.ForwardOnError(error);
		}

		public virtual void OnCompleted()
		{
			base.ForwardOnCompleted();
		}

		public IObserver<TTarget> GetForwarder()
		{
			return new Sink<TSource, TTarget>._(this);
		}

		private sealed class _ : IObserver<TTarget>
		{
			public _(Sink<TSource, TTarget> forward)
			{
				this._forward = forward;
			}

			public void OnNext(TTarget value)
			{
				this._forward.ForwardOnNext(value);
			}

			public void OnError(Exception error)
			{
				this._forward.ForwardOnError(error);
			}

			public void OnCompleted()
			{
				this._forward.ForwardOnCompleted();
			}

			private readonly Sink<TSource, TTarget> _forward;
		}
	}
}
