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

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Finally<TSource> : Producer<TSource, Finally<TSource>._>
	{
		public Finally(IObservable<TSource> source, Action finallyAction)
		{
			this._source = source;
			this._finallyAction = finallyAction;
		}

		protected override Finally<TSource>._ CreateSink(IObserver<TSource> observer)
		{
			return new Finally<TSource>._(this._finallyAction, observer);
		}

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

		private readonly IObservable<TSource> _source;

		private readonly Action _finallyAction;

		internal sealed class _ : IdentitySink<TSource>
		{
			public _(Action finallyAction, IObserver<TSource> observer)
				: base(observer)
			{
				this._finallyAction = finallyAction;
			}

			public override void Run(IObservable<TSource> source)
			{
				IDisposable disposable = source.SubscribeSafe(this);
				if (Interlocked.CompareExchange<IDisposable>(ref this._sourceDisposable, disposable, null) == BooleanDisposable.True)
				{
					try
					{
						disposable.Dispose();
					}
					finally
					{
						this._finallyAction();
					}
				}
			}

			protected override void Dispose(bool disposing)
			{
				base.Dispose(disposing);
				if (disposing)
				{
					IDisposable disposable = Interlocked.Exchange<IDisposable>(ref this._sourceDisposable, BooleanDisposable.True);
					if (disposable != BooleanDisposable.True && disposable != null)
					{
						try
						{
							disposable.Dispose();
						}
						finally
						{
							this._finallyAction();
						}
					}
				}
			}

			private readonly Action _finallyAction;

			private IDisposable _sourceDisposable;
		}
	}
}
