using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Reactive.Disposables;
using System.Threading;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class GetEnumerator<TSource> : IEnumerator<TSource>, IDisposable, IEnumerator, IObserver<TSource>
	{
		public GetEnumerator()
		{
			this._queue = new ConcurrentQueue<TSource>();
			this._gate = new SemaphoreSlim(0);
		}

		public IEnumerator<TSource> Run(IObservable<TSource> source)
		{
			Disposable.TrySetSingle(ref this._subscription, source.Subscribe(this));
			return this;
		}

		public void OnNext(TSource value)
		{
			this._queue.Enqueue(value);
			this._gate.Release();
		}

		public void OnError(Exception error)
		{
			this._error = error;
			Disposable.TryDispose(ref this._subscription);
			this._gate.Release();
		}

		public void OnCompleted()
		{
			this._done = true;
			Disposable.TryDispose(ref this._subscription);
			this._gate.Release();
		}

		public bool MoveNext()
		{
			this._gate.Wait();
			if (this._disposed)
			{
				throw new ObjectDisposedException("");
			}
			if (this._queue.TryDequeue(out this._current))
			{
				return true;
			}
			this._error.ThrowIfNotNull();
			this._gate.Release();
			return false;
		}

		public TSource Current
		{
			get
			{
				return this._current;
			}
		}

		object IEnumerator.Current
		{
			get
			{
				return this._current;
			}
		}

		public void Dispose()
		{
			Disposable.TryDispose(ref this._subscription);
			this._disposed = true;
			this._gate.Release();
		}

		public void Reset()
		{
			throw new NotSupportedException();
		}

		private readonly ConcurrentQueue<TSource> _queue;

		private TSource _current;

		private Exception _error;

		private bool _done;

		private bool _disposed;

		private IDisposable _subscription;

		private readonly SemaphoreSlim _gate;
	}
}
