using System;
using System.Reactive.Concurrency;

namespace System.Reactive
{
	public abstract class ObservableBase<T> : IObservable<T>
	{
		public IDisposable Subscribe(IObserver<T> observer)
		{
			if (observer == null)
			{
				throw new ArgumentNullException("observer");
			}
			AutoDetachObserver<T> autoDetachObserver = new AutoDetachObserver<T>(observer);
			if (CurrentThreadScheduler.IsScheduleRequired)
			{
				CurrentThreadScheduler.Instance.ScheduleAction(autoDetachObserver, new Action<AutoDetachObserver<T>>(this.ScheduledSubscribe));
			}
			else
			{
				try
				{
					autoDetachObserver.SetResource(this.SubscribeCore(autoDetachObserver));
				}
				catch (Exception ex)
				{
					if (!autoDetachObserver.Fail(ex))
					{
						throw;
					}
				}
			}
			return autoDetachObserver;
		}

		private void ScheduledSubscribe(AutoDetachObserver<T> autoDetachObserver)
		{
			try
			{
				autoDetachObserver.SetResource(this.SubscribeCore(autoDetachObserver));
			}
			catch (Exception ex)
			{
				if (!autoDetachObserver.Fail(ex))
				{
					throw;
				}
			}
		}

		protected abstract IDisposable SubscribeCore(IObserver<T> observer);
	}
}
