using System;
using System.Reactive.Disposables;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Using<TSource, T> : Producer<TSource, Using<TSource, T>._> where T : IDisposable
	{
		public Using(Func<T> resourceFactory, Func<T, IObservable<TSource>> observableFactory)
		{
			this._resourceFactory = resourceFactory;
			this._observableFactory = observableFactory;
		}

		protected override Using<TSource, T>._ CreateSink(IObserver<TSource> observer)
		{
			return new Using<TSource, T>._(observer);
		}

		protected override void Run(Using<TSource, T>._ sink)
		{
			sink.Run(this);
		}

		private readonly Func<T> _resourceFactory;

		private readonly Func<T, IObservable<TSource>> _observableFactory;

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

			public void Run(Using<TSource, T> parent)
			{
				IObservable<TSource> observable = null;
				IDisposable disposable = Disposable.Empty;
				try
				{
					T t = parent._resourceFactory();
					if (t != null)
					{
						disposable = t;
					}
					observable = parent._observableFactory(t);
				}
				catch (Exception ex)
				{
					observable = Observable.Throw<TSource>(ex);
				}
				this.Run(observable);
				Disposable.SetSingle(ref this._disposable, disposable);
			}

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

			private IDisposable _disposable;
		}
	}
}
