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

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Timeout<TSource, TTimeout> : Producer<TSource, Timeout<TSource, TTimeout>._>
	{
		public Timeout(IObservable<TSource> source, IObservable<TTimeout> firstTimeout, Func<TSource, IObservable<TTimeout>> timeoutSelector, IObservable<TSource> other)
		{
			this._source = source;
			this._firstTimeout = firstTimeout;
			this._timeoutSelector = timeoutSelector;
			this._other = other;
		}

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

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

		private readonly IObservable<TSource> _source;

		private readonly IObservable<TTimeout> _firstTimeout;

		private readonly Func<TSource, IObservable<TTimeout>> _timeoutSelector;

		private readonly IObservable<TSource> _other;

		internal sealed class _ : IdentitySink<TSource>
		{
			public _(Timeout<TSource, TTimeout> parent, IObserver<TSource> observer)
				: base(observer)
			{
				this._timeoutSelector = parent._timeoutSelector;
				this._other = parent._other;
			}

			public void Run(Timeout<TSource, TTimeout> parent)
			{
				this.SetTimer(parent._firstTimeout, 0L);
				Disposable.TrySetSingle(ref this._sourceDisposable, parent._source.SubscribeSafe(this));
			}

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

			public override void OnNext(TSource value)
			{
				long num = Volatile.Read(ref this._index);
				if (num != 9223372036854775807L && Interlocked.CompareExchange(ref this._index, num + 1L, num) == num)
				{
					IDisposable disposable = Volatile.Read<IDisposable>(ref this._timerDisposable);
					if (disposable != null)
					{
						disposable.Dispose();
					}
					base.ForwardOnNext(value);
					IObservable<TTimeout> observable = null;
					try
					{
						observable = this._timeoutSelector(value);
					}
					catch (Exception ex)
					{
						base.ForwardOnError(ex);
						return;
					}
					this.SetTimer(observable, num + 1L);
				}
			}

			public override void OnError(Exception error)
			{
				if (Interlocked.Exchange(ref this._index, 9223372036854775807L) != 9223372036854775807L)
				{
					base.ForwardOnError(error);
				}
			}

			public override void OnCompleted()
			{
				if (Interlocked.Exchange(ref this._index, 9223372036854775807L) != 9223372036854775807L)
				{
					base.ForwardOnCompleted();
				}
			}

			private void Timeout(long idx)
			{
				if (Volatile.Read(ref this._index) == idx && Interlocked.CompareExchange(ref this._index, 9223372036854775807L, idx) == idx)
				{
					Disposable.TrySetSerial(ref this._sourceDisposable, this._other.SubscribeSafe(base.GetForwarder()));
				}
			}

			private bool TimeoutError(long idx, Exception error)
			{
				if (Volatile.Read(ref this._index) == idx && Interlocked.CompareExchange(ref this._index, 9223372036854775807L, idx) == idx)
				{
					base.ForwardOnError(error);
					return true;
				}
				return false;
			}

			private void SetTimer(IObservable<TTimeout> timeout, long idx)
			{
				Timeout<TSource, TTimeout>._.TimeoutObserver timeoutObserver = new Timeout<TSource, TTimeout>._.TimeoutObserver(this, idx);
				if (Disposable.TrySetSerial(ref this._timerDisposable, timeoutObserver))
				{
					IDisposable disposable = timeout.Subscribe(timeoutObserver);
					timeoutObserver.SetResource(disposable);
				}
			}

			private readonly Func<TSource, IObservable<TTimeout>> _timeoutSelector;

			private readonly IObservable<TSource> _other;

			private IDisposable _sourceDisposable;

			private IDisposable _timerDisposable;

			private long _index;

			private sealed class TimeoutObserver : SafeObserver<TTimeout>
			{
				public TimeoutObserver(Timeout<TSource, TTimeout>._ parent, long id)
				{
					this._parent = parent;
					this._id = id;
				}

				public override void OnNext(TTimeout value)
				{
					this.OnCompleted();
				}

				public override void OnError(Exception error)
				{
					if (!this._parent.TimeoutError(this._id, error))
					{
						base.Dispose();
					}
				}

				public override void OnCompleted()
				{
					this._parent.Timeout(this._id);
					base.Dispose();
				}

				private readonly Timeout<TSource, TTimeout>._ _parent;

				private readonly long _id;
			}
		}
	}
}
