using System;
using System.Threading;

namespace System.Reactive.Concurrency
{
	internal static class ObserveOn<TSource>
	{
		internal sealed class Scheduler : Producer<TSource, ObserveOnObserverNew<TSource>>
		{
			public Scheduler(IObservable<TSource> source, IScheduler scheduler)
			{
				this._source = source;
				this._scheduler = scheduler;
			}

			protected override ObserveOnObserverNew<TSource> CreateSink(IObserver<TSource> observer)
			{
				return new ObserveOnObserverNew<TSource>(this._scheduler, observer);
			}

			protected override void Run(ObserveOnObserverNew<TSource> sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly IScheduler _scheduler;
		}

		internal sealed class SchedulerLongRunning : Producer<TSource, ObserveOnObserverLongRunning<TSource>>
		{
			public SchedulerLongRunning(IObservable<TSource> source, ISchedulerLongRunning scheduler)
			{
				this._source = source;
				this._scheduler = scheduler;
			}

			protected override ObserveOnObserverLongRunning<TSource> CreateSink(IObserver<TSource> observer)
			{
				return new ObserveOnObserverLongRunning<TSource>(this._scheduler, observer);
			}

			protected override void Run(ObserveOnObserverLongRunning<TSource> sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly ISchedulerLongRunning _scheduler;
		}

		internal sealed class Context : Producer<TSource, ObserveOn<TSource>.Context._>
		{
			public Context(IObservable<TSource> source, SynchronizationContext context)
			{
				this._source = source;
				this._context = context;
			}

			protected override ObserveOn<TSource>.Context._ CreateSink(IObserver<TSource> observer)
			{
				return new ObserveOn<TSource>.Context._(this._context, observer);
			}

			protected override void Run(ObserveOn<TSource>.Context._ sink)
			{
				sink.Run(this._source);
			}

			private readonly IObservable<TSource> _source;

			private readonly SynchronizationContext _context;

			internal sealed class _ : IdentitySink<TSource>
			{
				public _(SynchronizationContext context, IObserver<TSource> observer)
					: base(observer)
				{
					this._context = context;
				}

				public override void Run(IObservable<TSource> source)
				{
					this._context.OperationStarted();
					base.SetUpstream(source.SubscribeSafe(this));
				}

				protected override void Dispose(bool disposing)
				{
					if (disposing)
					{
						this._context.OperationCompleted();
					}
					base.Dispose(disposing);
				}

				public override void OnNext(TSource value)
				{
					this._context.Post(new SendOrPostCallback(this.OnNextPosted), value);
				}

				public override void OnError(Exception error)
				{
					this._context.Post(new SendOrPostCallback(this.OnErrorPosted), error);
				}

				public override void OnCompleted()
				{
					this._context.Post(new SendOrPostCallback(this.OnCompletedPosted), null);
				}

				private void OnNextPosted(object value)
				{
					base.ForwardOnNext((TSource)((object)value));
				}

				private void OnErrorPosted(object error)
				{
					base.ForwardOnError((Exception)error);
				}

				private void OnCompletedPosted(object ignored)
				{
					base.ForwardOnCompleted();
				}

				private readonly SynchronizationContext _context;
			}
		}
	}
}
