using System;
using System.Reactive.Disposables;
using System.Threading;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Catch<TSource, T> : Producer<TSource, Catch<TSource, T>._> where T : Exception
	{
		public Catch(IObservable<TSource> source, Func<T, IObservable<TSource>> handler)
		{
			this._source = source;
			this._handler = handler;
		}

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

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

		private readonly IObservable<TSource> _source;

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

		internal sealed class _ : IdentitySink<TSource>
		{
			public _(Func<T, IObservable<TSource>> handler, IObserver<TSource> observer)
				: base(observer)
			{
				this._handler = handler;
			}

			public override void Run(IObservable<TSource> source)
			{
				Disposable.TrySetSingle(ref this._subscription, source.SubscribeSafe(this));
			}

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

			public override void OnError(Exception error)
			{
				if (!Volatile.Read(ref this._once))
				{
					T t = error as T;
					if (t != null)
					{
						IObservable<TSource> observable = null;
						try
						{
							observable = this._handler(t);
						}
						catch (Exception ex)
						{
							base.ForwardOnError(ex);
							return;
						}
						Volatile.Write(ref this._once, true);
						Disposable.TrySetSerial(ref this._subscription, observable.SubscribeSafe(this));
						return;
					}
				}
				base.ForwardOnError(error);
			}

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

			private bool _once;

			private IDisposable _subscription;
		}
	}
}
