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

namespace System.Reactive
{
	internal sealed class ObserveOnObserver<T> : ScheduledObserver<T>
	{
		public ObserveOnObserver(IScheduler scheduler, IObserver<T> observer)
			: base(scheduler, observer)
		{
		}

		public void Run(IObservable<T> source)
		{
			Disposable.SetSingle(ref this._run, source.SubscribeSafe(this));
		}

		protected override void OnNextCore(T value)
		{
			base.OnNextCore(value);
			base.EnsureActive();
		}

		protected override void OnErrorCore(Exception exception)
		{
			base.OnErrorCore(exception);
			base.EnsureActive();
		}

		protected override void OnCompletedCore()
		{
			base.OnCompletedCore();
			base.EnsureActive();
		}

		protected override void Dispose(bool disposing)
		{
			base.Dispose(disposing);
			if (disposing)
			{
				Disposable.TryDispose(ref this._run);
			}
		}

		private IDisposable _run;
	}
}
