using System;
using System.Reactive.Concurrency;
using System.Runtime.CompilerServices;

namespace System.Reactive
{
	internal abstract class Producer<TTarget, TSink> : IProducer<TTarget>, IObservable<TTarget> where TSink : IDisposable
	{
		public IDisposable Subscribe(IObserver<TTarget> observer)
		{
			if (observer == null)
			{
				throw new ArgumentNullException("observer");
			}
			return this.SubscribeRaw(observer, true);
		}

		public IDisposable SubscribeRaw(IObserver<TTarget> observer, bool enableSafeguard)
		{
			ISafeObserver<TTarget> safeObserver = null;
			if (enableSafeguard)
			{
				safeObserver = (observer = SafeObserver<TTarget>.Wrap(observer));
			}
			TSink tsink = this.CreateSink(observer);
			if (safeObserver != null)
			{
				safeObserver.SetResource(tsink);
			}
			if (CurrentThreadScheduler.IsScheduleRequired)
			{
				CurrentThreadScheduler.Instance.ScheduleAction(new ValueTuple<Producer<TTarget, TSink>, TSink>(this, tsink), delegate([TupleElementNames(new string[] { "this", "sink" })] ValueTuple<Producer<TTarget, TSink>, TSink> tuple)
				{
					tuple.Item1.Run(tuple.Item2);
				});
			}
			else
			{
				this.Run(tsink);
			}
			return tsink;
		}

		protected abstract void Run(TSink sink);

		protected abstract TSink CreateSink(IObserver<TTarget> observer);
	}
}
