using System;
using System.Reactive.Disposables;
using System.Threading;

namespace System.Reactive
{
	internal abstract class Sink<TTarget> : ISink<TTarget>, IDisposable
	{
		protected Sink(IObserver<TTarget> observer)
		{
			this._observer = observer;
		}

		public void Dispose()
		{
			if (Interlocked.Exchange<IObserver<TTarget>>(ref this._observer, NopObserver<TTarget>.Instance) != NopObserver<TTarget>.Instance)
			{
				this.Dispose(true);
			}
		}

		protected virtual void Dispose(bool disposing)
		{
			Disposable.TryDispose(ref this._upstream);
		}

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

		public void ForwardOnCompleted()
		{
			this._observer.OnCompleted();
			this.Dispose();
		}

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

		protected void SetUpstream(IDisposable upstream)
		{
			Disposable.SetSingle(ref this._upstream, upstream);
		}

		protected void DisposeUpstream()
		{
			Disposable.TryDispose(ref this._upstream);
		}

		private IDisposable _upstream;

		private volatile IObserver<TTarget> _observer;
	}
}
