using System;
using System.Reactive.Disposables;
using System.Reactive.Subjects;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Multicast<TSource, TIntermediate, TResult> : Producer<TResult, Multicast<TSource, TIntermediate, TResult>._>
	{
		public Multicast(IObservable<TSource> source, Func<ISubject<TSource, TIntermediate>> subjectSelector, Func<IObservable<TIntermediate>, IObservable<TResult>> selector)
		{
			this._source = source;
			this._subjectSelector = subjectSelector;
			this._selector = selector;
		}

		protected override Multicast<TSource, TIntermediate, TResult>._ CreateSink(IObserver<TResult> observer)
		{
			return new Multicast<TSource, TIntermediate, TResult>._(observer);
		}

		protected override void Run(Multicast<TSource, TIntermediate, TResult>._ sink)
		{
			sink.Run(this);
		}

		private readonly IObservable<TSource> _source;

		private readonly Func<ISubject<TSource, TIntermediate>> _subjectSelector;

		private readonly Func<IObservable<TIntermediate>, IObservable<TResult>> _selector;

		internal sealed class _ : IdentitySink<TResult>
		{
			public _(IObserver<TResult> observer)
				: base(observer)
			{
			}

			public void Run(Multicast<TSource, TIntermediate, TResult> parent)
			{
				IObservable<TResult> observable = null;
				IConnectableObservable<TIntermediate> connectableObservable = null;
				try
				{
					ISubject<TSource, TIntermediate> subject = parent._subjectSelector();
					connectableObservable = new ConnectableObservable<TSource, TIntermediate>(parent._source, subject);
					observable = parent._selector(connectableObservable);
				}
				catch (Exception ex)
				{
					base.ForwardOnError(ex);
					return;
				}
				this.Run(observable);
				Disposable.SetSingle(ref this._connection, connectableObservable.Connect());
			}

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

			private IDisposable _connection;
		}
	}
}
