using System;
using System.Collections.Generic;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;

namespace System.Reactive.Subjects
{
	public sealed class ReplaySubject<T> : SubjectBase<T>
	{
		public ReplaySubject()
			: this(int.MaxValue)
		{
		}

		public ReplaySubject(IScheduler scheduler)
		{
			this._implementation = new ReplaySubject<T>.ReplayByTime(scheduler);
		}

		public ReplaySubject(int bufferSize)
		{
			if (bufferSize == 1)
			{
				this._implementation = new ReplaySubject<T>.ReplayOne();
				return;
			}
			if (bufferSize != 2147483647)
			{
				this._implementation = new ReplaySubject<T>.ReplayMany(bufferSize);
				return;
			}
			this._implementation = new ReplaySubject<T>.ReplayAll();
		}

		public ReplaySubject(int bufferSize, IScheduler scheduler)
		{
			this._implementation = new ReplaySubject<T>.ReplayByTime(bufferSize, scheduler);
		}

		public ReplaySubject(TimeSpan window)
		{
			this._implementation = new ReplaySubject<T>.ReplayByTime(window);
		}

		public ReplaySubject(TimeSpan window, IScheduler scheduler)
		{
			this._implementation = new ReplaySubject<T>.ReplayByTime(window, scheduler);
		}

		public ReplaySubject(int bufferSize, TimeSpan window)
		{
			this._implementation = new ReplaySubject<T>.ReplayByTime(bufferSize, window);
		}

		public ReplaySubject(int bufferSize, TimeSpan window, IScheduler scheduler)
		{
			this._implementation = new ReplaySubject<T>.ReplayByTime(bufferSize, window, scheduler);
		}

		public override bool HasObservers
		{
			get
			{
				return this._implementation.HasObservers;
			}
		}

		public override bool IsDisposed
		{
			get
			{
				return this._implementation.IsDisposed;
			}
		}

		public override void OnNext(T value)
		{
			this._implementation.OnNext(value);
		}

		public override void OnError(Exception error)
		{
			if (error == null)
			{
				throw new ArgumentNullException("error");
			}
			this._implementation.OnError(error);
		}

		public override void OnCompleted()
		{
			this._implementation.OnCompleted();
		}

		public override IDisposable Subscribe(IObserver<T> observer)
		{
			if (observer == null)
			{
				throw new ArgumentNullException("observer");
			}
			return this._implementation.Subscribe(observer);
		}

		public override void Dispose()
		{
			this._implementation.Dispose();
		}

		private readonly SubjectBase<T> _implementation;

		private abstract class ReplayBase : SubjectBase<T>
		{
			protected ReplayBase()
			{
				this._observers = ImmutableList<Interface3<T>>.Empty;
				this._isStopped = false;
				this._error = null;
			}

			public override bool HasObservers
			{
				get
				{
					ImmutableList<Interface3<T>> observers = this._observers;
					return observers != null && observers.Data.Length != 0;
				}
			}

			public override bool IsDisposed
			{
				get
				{
					object gate = this._gate;
					bool isDisposed;
					lock (gate)
					{
						isDisposed = this._isDisposed;
					}
					return isDisposed;
				}
			}

			public override void OnNext(T value)
			{
				Interface3<T>[] array = null;
				object gate = this._gate;
				lock (gate)
				{
					this.CheckDisposed();
					if (!this._isStopped)
					{
						this.Next(value);
						this.Trim();
						array = this._observers.Data;
						Interface3<T>[] array2 = array;
						for (int i = 0; i < array2.Length; i++)
						{
							array2[i].OnNext(value);
						}
					}
				}
				if (array != null)
				{
					Interface3<T>[] array2 = array;
					for (int i = 0; i < array2.Length; i++)
					{
						array2[i].EnsureActive();
					}
				}
			}

			public override void OnError(Exception error)
			{
				Interface3<T>[] array = null;
				object gate = this._gate;
				lock (gate)
				{
					this.CheckDisposed();
					if (!this._isStopped)
					{
						this._isStopped = true;
						this._error = error;
						this.Trim();
						array = this._observers.Data;
						Interface3<T>[] array2 = array;
						for (int i = 0; i < array2.Length; i++)
						{
							array2[i].OnError(error);
						}
						this._observers = ImmutableList<Interface3<T>>.Empty;
					}
				}
				if (array != null)
				{
					Interface3<T>[] array2 = array;
					for (int i = 0; i < array2.Length; i++)
					{
						array2[i].EnsureActive();
					}
				}
			}

			public override void OnCompleted()
			{
				Interface3<T>[] array = null;
				object gate = this._gate;
				lock (gate)
				{
					this.CheckDisposed();
					if (!this._isStopped)
					{
						this._isStopped = true;
						this.Trim();
						array = this._observers.Data;
						Interface3<T>[] array2 = array;
						for (int i = 0; i < array2.Length; i++)
						{
							array2[i].OnCompleted();
						}
						this._observers = ImmutableList<Interface3<T>>.Empty;
					}
				}
				if (array != null)
				{
					Interface3<T>[] array2 = array;
					for (int i = 0; i < array2.Length; i++)
					{
						array2[i].EnsureActive();
					}
				}
			}

			public override IDisposable Subscribe(IObserver<T> observer)
			{
				Interface3<T> @interface = this.CreateScheduledObserver(observer);
				int num = 0;
				IDisposable disposable = Disposable.Empty;
				object gate = this._gate;
				lock (gate)
				{
					this.CheckDisposed();
					this.Trim();
					num = this.Replay(@interface);
					if (this._error != null)
					{
						num++;
						@interface.OnError(this._error);
					}
					else if (this._isStopped)
					{
						num++;
						@interface.OnCompleted();
					}
					if (!this._isStopped)
					{
						disposable = new ReplaySubject<T>.ReplayBase.Subscription(this, @interface);
						this._observers = this._observers.Add(@interface);
					}
				}
				@interface.EnsureActive(num);
				return disposable;
			}

			public override void Dispose()
			{
				object gate = this._gate;
				lock (gate)
				{
					this._isDisposed = true;
					this._observers = null;
					this.DisposeCore();
				}
			}

			protected abstract void DisposeCore();

			protected abstract void Next(T value);

			protected abstract int Replay(IObserver<T> observer);

			protected abstract void Trim();

			protected abstract Interface3<T> CreateScheduledObserver(IObserver<T> observer);

			private void CheckDisposed()
			{
				if (this._isDisposed)
				{
					throw new ObjectDisposedException(string.Empty);
				}
			}

			private void Unsubscribe(Interface3<T> observer)
			{
				object gate = this._gate;
				lock (gate)
				{
					if (!this._isDisposed)
					{
						this._observers = this._observers.Remove(observer);
					}
				}
			}

			private readonly object _gate = new object();

			private ImmutableList<Interface3<T>> _observers;

			private bool _isStopped;

			private Exception _error;

			private bool _isDisposed;

			private sealed class Subscription : IDisposable
			{
				public Subscription(ReplaySubject<T>.ReplayBase subject, Interface3<T> observer)
				{
					this._subject = subject;
					this._observer = observer;
				}

				public void Dispose()
				{
					this._observer.Dispose();
					this._subject.Unsubscribe(this._observer);
				}

				private readonly ReplaySubject<T>.ReplayBase _subject;

				private readonly Interface3<T> _observer;
			}
		}

		private sealed class ReplayByTime : ReplaySubject<T>.ReplayBase
		{
			public ReplayByTime(int bufferSize, TimeSpan window, IScheduler scheduler)
			{
				if (bufferSize < 0)
				{
					throw new ArgumentOutOfRangeException("bufferSize");
				}
				if (window < TimeSpan.Zero)
				{
					throw new ArgumentOutOfRangeException("window");
				}
				this._bufferSize = bufferSize;
				this._window = window;
				if (scheduler == null)
				{
					throw new ArgumentNullException("scheduler");
				}
				this._scheduler = scheduler;
				this._stopwatch = this._scheduler.StartStopwatch();
				this._queue = new Queue<TimeInterval<T>>();
			}

			public ReplayByTime(int bufferSize, TimeSpan window)
				: this(bufferSize, window, SchedulerDefaults.Iteration)
			{
			}

			public ReplayByTime(IScheduler scheduler)
				: this(int.MaxValue, TimeSpan.MaxValue, scheduler)
			{
			}

			public ReplayByTime(int bufferSize, IScheduler scheduler)
				: this(bufferSize, TimeSpan.MaxValue, scheduler)
			{
			}

			public ReplayByTime(TimeSpan window, IScheduler scheduler)
				: this(int.MaxValue, window, scheduler)
			{
			}

			public ReplayByTime(TimeSpan window)
				: this(int.MaxValue, window, SchedulerDefaults.Iteration)
			{
			}

			protected override Interface3<T> CreateScheduledObserver(IObserver<T> observer)
			{
				return new ScheduledObserver<T>(this._scheduler, observer);
			}

			protected override void DisposeCore()
			{
				this._queue.Clear();
			}

			protected override void Next(T value)
			{
				TimeSpan elapsed = this._stopwatch.Elapsed;
				this._queue.Enqueue(new TimeInterval<T>(value, elapsed));
			}

			protected override int Replay(IObserver<T> observer)
			{
				int count = this._queue.Count;
				foreach (TimeInterval<T> timeInterval in this._queue)
				{
					observer.OnNext(timeInterval.Value);
				}
				return count;
			}

			protected override void Trim()
			{
				TimeSpan elapsed = this._stopwatch.Elapsed;
				while (this._queue.Count > this._bufferSize)
				{
					this._queue.Dequeue();
				}
				while (this._queue.Count > 0 && elapsed.Subtract(this._queue.Peek().Interval).CompareTo(this._window) > 0)
				{
					this._queue.Dequeue();
				}
			}

			private readonly int _bufferSize;

			private readonly TimeSpan _window;

			private readonly IScheduler _scheduler;

			private readonly IStopwatch _stopwatch;

			private readonly Queue<TimeInterval<T>> _queue;
		}

		private sealed class ReplayOne : ReplaySubject<T>.ReplayBufferBase
		{
			protected override void Trim()
			{
			}

			protected override void Next(T value)
			{
				this._hasValue = true;
				this._value = value;
			}

			protected override int Replay(IObserver<T> observer)
			{
				int num = 0;
				if (this._hasValue)
				{
					num = 1;
					observer.OnNext(this._value);
				}
				return num;
			}

			protected override void DisposeCore()
			{
				this._value = default(T);
			}

			private bool _hasValue;

			private T _value;
		}

		private sealed class ReplayMany : ReplaySubject<T>.ReplayManyBase
		{
			public ReplayMany(int bufferSize)
				: base(bufferSize)
			{
				this._bufferSize = bufferSize;
			}

			protected override void Trim()
			{
				while (this._queue.Count > this._bufferSize)
				{
					this._queue.Dequeue();
				}
			}

			private readonly int _bufferSize;
		}

		private sealed class ReplayAll : ReplaySubject<T>.ReplayManyBase
		{
			public ReplayAll()
				: base(0)
			{
			}

			protected override void Trim()
			{
			}
		}

		private abstract class ReplayBufferBase : ReplaySubject<T>.ReplayBase
		{
			protected override Interface3<T> CreateScheduledObserver(IObserver<T> observer)
			{
				return new FastImmediateObserver<T>(observer);
			}

			protected override void DisposeCore()
			{
			}
		}

		private abstract class ReplayManyBase : ReplaySubject<T>.ReplayBufferBase
		{
			protected ReplayManyBase(int queueSize)
			{
				this._queue = new Queue<T>(Math.Min(queueSize, 64));
			}

			protected override void Next(T value)
			{
				this._queue.Enqueue(value);
			}

			protected override int Replay(IObserver<T> observer)
			{
				int count = this._queue.Count;
				foreach (T t in this._queue)
				{
					observer.OnNext(t);
				}
				return count;
			}

			protected override void DisposeCore()
			{
				this._queue.Clear();
			}

			protected readonly Queue<T> _queue;
		}
	}
}
