using System;
using System.ComponentModel;
using System.Reactive.Disposables;
using System.Runtime.CompilerServices;
using System.Threading;

namespace System.Reactive.Concurrency
{
	[EditorBrowsable(EditorBrowsableState.Advanced)]
	public static class Synchronization
	{
		public static IObservable<TSource> SubscribeOn<TSource>(IObservable<TSource> source, IScheduler scheduler)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			if (scheduler == null)
			{
				throw new ArgumentNullException("scheduler");
			}
			return new Synchronization.SubscribeOnObservable<TSource>(source, scheduler);
		}

		public static IObservable<TSource> SubscribeOn<TSource>(IObservable<TSource> source, SynchronizationContext context)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			if (context == null)
			{
				throw new ArgumentNullException("context");
			}
			return new Synchronization.SubscribeOnCtxObservable<TSource>(source, context);
		}

		public static IObservable<TSource> ObserveOn<TSource>(IObservable<TSource> source, IScheduler scheduler)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			if (scheduler == null)
			{
				throw new ArgumentNullException("scheduler");
			}
			ISchedulerLongRunning schedulerLongRunning = scheduler.AsLongRunning();
			if (schedulerLongRunning != null)
			{
				return new ObserveOn<TSource>.SchedulerLongRunning(source, schedulerLongRunning);
			}
			return new ObserveOn<TSource>.Scheduler(source, scheduler);
		}

		public static IObservable<TSource> ObserveOn<TSource>(IObservable<TSource> source, SynchronizationContext context)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			if (context == null)
			{
				throw new ArgumentNullException("context");
			}
			return new ObserveOn<TSource>.Context(source, context);
		}

		public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> source)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			return new Synchronize<TSource>(source);
		}

		public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> source, object gate)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			if (gate == null)
			{
				throw new ArgumentNullException("gate");
			}
			return new Synchronize<TSource>(source, gate);
		}

		private sealed class SubscribeOnObservable<TSource> : ObservableBase<TSource>
		{
			public SubscribeOnObservable(IObservable<TSource> source, IScheduler scheduler)
			{
				this._source = source;
				this._scheduler = scheduler;
			}

			protected override IDisposable SubscribeCore(IObserver<TSource> observer)
			{
				return new Synchronization.SubscribeOnObservable<TSource>.Subscription(this._source, this._scheduler, observer);
			}

			private readonly IObservable<TSource> _source;

			private readonly IScheduler _scheduler;

			private sealed class Subscription : IDisposable
			{
				public Subscription(IObservable<TSource> source, IScheduler scheduler, IObserver<TSource> observer)
				{
					Disposable.TrySetSingle(ref this._cancel, scheduler.Schedule<ValueTuple<Synchronization.SubscribeOnObservable<TSource>.Subscription, IObservable<TSource>, IObserver<TSource>>>(new ValueTuple<Synchronization.SubscribeOnObservable<TSource>.Subscription, IObservable<TSource>, IObserver<TSource>>(this, source, observer), delegate(IScheduler closureScheduler, [TupleElementNames(new string[] { "this", "source", "observer" })] ValueTuple<Synchronization.SubscribeOnObservable<TSource>.Subscription, IObservable<TSource>, IObserver<TSource>> state)
					{
						Disposable.TrySetSerial(ref state.Item1._cancel, new ScheduledDisposable(closureScheduler, state.Item2.SubscribeSafe(state.Item3)));
						return Disposable.Empty;
					}));
				}

				public void Dispose()
				{
					Disposable.TryDispose(ref this._cancel);
				}

				private IDisposable _cancel;
			}
		}

		private sealed class SubscribeOnCtxObservable<TSource> : ObservableBase<TSource>
		{
			public SubscribeOnCtxObservable(IObservable<TSource> source, SynchronizationContext context)
			{
				this._source = source;
				this._context = context;
			}

			protected override IDisposable SubscribeCore(IObserver<TSource> observer)
			{
				return new Synchronization.SubscribeOnCtxObservable<TSource>.Subscription(this._source, this._context, observer);
			}

			private readonly IObservable<TSource> _source;

			private readonly SynchronizationContext _context;

			private sealed class Subscription : IDisposable
			{
				public Subscription(IObservable<TSource> source, SynchronizationContext context, IObserver<TSource> observer)
				{
					this._source = source;
					this._context = context;
					this._observer = observer;
					context.PostWithStartComplete(delegate(Synchronization.SubscribeOnCtxObservable<TSource>.Subscription @this)
					{
						if (!Disposable.GetIsDisposed(ref @this._cancel))
						{
							Disposable.SetSingle(ref @this._cancel, new ContextDisposable(@this._context, @this._source.SubscribeSafe(@this._observer)));
						}
					}, this);
				}

				public void Dispose()
				{
					Disposable.TryDispose(ref this._cancel);
				}

				private readonly IObservable<TSource> _source;

				private readonly IObserver<TSource> _observer;

				private readonly SynchronizationContext _context;

				private IDisposable _cancel;
			}
		}
	}
}
