using System;

namespace System.Reactive.Linq.ObservableImpl
{
	internal sealed class Collect<TSource, TResult> : PushToPullAdapter<TSource, TResult>
	{
		public Collect(IObservable<TSource> source, Func<TResult> getInitialCollector, Func<TResult, TSource, TResult> merge, Func<TResult, TResult> getNewCollector)
			: base(source)
		{
			this._getInitialCollector = getInitialCollector;
			this._merge = merge;
			this._getNewCollector = getNewCollector;
		}

		protected override PushToPullSink<TSource, TResult> Run()
		{
			return new Collect<TSource, TResult>._(this._merge, this._getNewCollector, this._getInitialCollector());
		}

		private readonly Func<TResult> _getInitialCollector;

		private readonly Func<TResult, TSource, TResult> _merge;

		private readonly Func<TResult, TResult> _getNewCollector;

		private sealed class _ : PushToPullSink<TSource, TResult>
		{
			public _(Func<TResult, TSource, TResult> merge, Func<TResult, TResult> getNewCollector, TResult collector)
			{
				this._gate = new object();
				this._merge = merge;
				this._getNewCollector = getNewCollector;
				this._collector = collector;
			}

			public override void OnNext(TSource value)
			{
				object gate = this._gate;
				lock (gate)
				{
					try
					{
						this._collector = this._merge(this._collector, value);
					}
					catch (Exception ex)
					{
						this._error = ex;
						base.Dispose();
					}
				}
			}

			public override void OnError(Exception error)
			{
				base.Dispose();
				object gate = this._gate;
				lock (gate)
				{
					this._error = error;
				}
			}

			public override void OnCompleted()
			{
				base.Dispose();
				object gate = this._gate;
				lock (gate)
				{
					this._hasCompleted = true;
				}
			}

			public override bool TryMoveNext(out TResult current)
			{
				object gate = this._gate;
				bool flag2;
				lock (gate)
				{
					Exception error = this._error;
					if (error != null)
					{
						current = default(TResult);
						this._collector = default(TResult);
						error.Throw();
					}
					else if (this._hasCompleted)
					{
						if (this._done)
						{
							current = default(TResult);
							this._collector = default(TResult);
							return false;
						}
						current = this._collector;
						this._done = true;
					}
					else
					{
						current = this._collector;
						try
						{
							this._collector = this._getNewCollector(current);
						}
						catch
						{
							base.Dispose();
							throw;
						}
					}
					flag2 = true;
				}
				return flag2;
			}

			private readonly object _gate;

			private readonly Func<TResult, TSource, TResult> _merge;

			private readonly Func<TResult, TResult> _getNewCollector;

			private TResult _collector;

			private Exception _error;

			private bool _hasCompleted;

			private bool _done;
		}
	}
}
