using System;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq.ObservableImpl;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

namespace System.Reactive.Threading.Tasks
{
	public static class TaskObservableExtensions
	{
		public static IObservable<Unit> ToObservable(this Task task)
		{
			if (task == null)
			{
				throw new ArgumentNullException("task");
			}
			return TaskObservableExtensions.ToObservableImpl(task, null);
		}

		public static IObservable<Unit> ToObservable(this Task task, IScheduler scheduler)
		{
			if (task == null)
			{
				throw new ArgumentNullException("task");
			}
			if (scheduler == null)
			{
				throw new ArgumentNullException("scheduler");
			}
			return TaskObservableExtensions.ToObservableImpl(task, scheduler);
		}

		private static IObservable<Unit> ToObservableImpl(object task, object scheduler)
		{
			if (!task.IsCompleted)
			{
				return new TaskObservableExtensions.SlowTaskObservable(task, scheduler);
			}
			scheduler = scheduler ?? ImmediateScheduler.Instance;
			TaskStatus status = task.Status;
			if (status == TaskStatus.Canceled)
			{
				return new Throw<Unit>(new TaskCanceledException(task), scheduler);
			}
			if (status == TaskStatus.Faulted)
			{
				return new Throw<Unit>(task.Exception.InnerException, scheduler);
			}
			return new Return<Unit>(Unit.Default, scheduler);
		}

		private static void EmitTaskResult(this object task, IObserver<Unit> subject)
		{
			switch (task.Status)
			{
			case TaskStatus.RanToCompletion:
				subject.OnNext(Unit.Default);
				subject.OnCompleted();
				return;
			case TaskStatus.Canceled:
				subject.OnError(new TaskCanceledException(task));
				return;
			case TaskStatus.Faulted:
				subject.OnError(task.Exception.InnerException);
				return;
			default:
				return;
			}
		}

		internal static IDisposable Subscribe(this Task task, IObserver<Unit> observer)
		{
			if (task.IsCompleted)
			{
				task.EmitTaskResult(observer);
				return Disposable.Empty;
			}
			CancellationDisposable cancellationDisposable = new CancellationDisposable();
			task.ContinueWith(delegate(Task t, object observerObject)
			{
				t.EmitTaskResult((IObserver<Unit>)observerObject);
			}, observer, cancellationDisposable.Token, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Current);
			return cancellationDisposable;
		}

		public static IObservable<TResult> ToObservable<TResult>(this Task<TResult> task)
		{
			if (task == null)
			{
				throw new ArgumentNullException("task");
			}
			return TaskObservableExtensions.ToObservableImpl<TResult>(task, null);
		}

		public static IObservable<TResult> ToObservable<TResult>(this Task<TResult> task, IScheduler scheduler)
		{
			if (task == null)
			{
				throw new ArgumentNullException("task");
			}
			if (scheduler == null)
			{
				throw new ArgumentNullException("scheduler");
			}
			return TaskObservableExtensions.ToObservableImpl<TResult>(task, scheduler);
		}

		private static IObservable<TResult> ToObservableImpl<TResult>(Task<TResult> task, object scheduler)
		{
			if (!task.IsCompleted)
			{
				return new TaskObservableExtensions.SlowTaskObservable<TResult>(task, scheduler);
			}
			scheduler = scheduler ?? ImmediateScheduler.Instance;
			TaskStatus status = task.Status;
			if (status == TaskStatus.Canceled)
			{
				return new Throw<TResult>(new TaskCanceledException(task), scheduler);
			}
			if (status == TaskStatus.Faulted)
			{
				return new Throw<TResult>(task.Exception.InnerException, scheduler);
			}
			return new Return<TResult>(task.Result, scheduler);
		}

		private static void EmitTaskResult<TResult>(this Task<TResult> task, IObserver<TResult> subject)
		{
			switch (task.Status)
			{
			case TaskStatus.RanToCompletion:
				subject.OnNext(task.Result);
				subject.OnCompleted();
				return;
			case TaskStatus.Canceled:
				subject.OnError(new TaskCanceledException(task));
				return;
			case TaskStatus.Faulted:
				subject.OnError(task.Exception.InnerException);
				return;
			default:
				return;
			}
		}

		private static TaskContinuationOptions GetTaskContinuationOptions(object scheduler)
		{
			TaskContinuationOptions taskContinuationOptions = TaskContinuationOptions.None;
			if (scheduler != null)
			{
				taskContinuationOptions |= TaskContinuationOptions.ExecuteSynchronously;
			}
			return taskContinuationOptions;
		}

		internal static IDisposable Subscribe<TResult>(this Task<TResult> task, IObserver<TResult> observer)
		{
			if (task.IsCompleted)
			{
				task.EmitTaskResult(observer);
				return Disposable.Empty;
			}
			CancellationDisposable cancellationDisposable = new CancellationDisposable();
			task.ContinueWith(delegate(Task<TResult> t, object observerObject)
			{
				t.EmitTaskResult((IObserver<TResult>)observerObject);
			}, observer, cancellationDisposable.Token, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Current);
			return cancellationDisposable;
		}

		public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable)
		{
			if (observable == null)
			{
				throw new ArgumentNullException("observable");
			}
			return observable.ToTask(default(CancellationToken), null);
		}

		public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable, object state)
		{
			if (observable == null)
			{
				throw new ArgumentNullException("observable");
			}
			return observable.ToTask(default(CancellationToken), state);
		}

		public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable, CancellationToken cancellationToken)
		{
			if (observable == null)
			{
				throw new ArgumentNullException("observable");
			}
			return observable.ToTask(cancellationToken, null);
		}

		public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable, CancellationToken cancellationToken, object state)
		{
			if (observable == null)
			{
				throw new ArgumentNullException("observable");
			}
			TaskCompletionSource<TResult> taskCompletionSource = new TaskCompletionSource<TResult>(state);
			TaskObservableExtensions.ToTaskObserver<TResult> toTaskObserver = new TaskObservableExtensions.ToTaskObserver<TResult>(taskCompletionSource, cancellationToken);
			try
			{
				toTaskObserver.SetResource(observable.Subscribe(toTaskObserver));
			}
			catch (Exception ex)
			{
				taskCompletionSource.TrySetException(ex);
			}
			return taskCompletionSource.Task;
		}

		private sealed class SlowTaskObservable : IObservable<Unit>
		{
			public SlowTaskObservable(Task task, IScheduler scheduler)
			{
				this._task = task;
				this._scheduler = scheduler;
			}

			public IDisposable Subscribe(IObserver<Unit> observer)
			{
				if (observer == null)
				{
					throw new ArgumentNullException("observer");
				}
				CancellationDisposable cancellationDisposable = new CancellationDisposable();
				TaskContinuationOptions taskContinuationOptions = TaskObservableExtensions.GetTaskContinuationOptions(this._scheduler);
				if (this._scheduler == null)
				{
					this._task.ContinueWith(delegate(Task t, object subjectObject)
					{
						t.EmitTaskResult((IObserver<Unit>)subjectObject);
					}, observer, cancellationDisposable.Token, taskContinuationOptions, TaskScheduler.Current);
				}
				else
				{
					this._task.ContinueWithState(delegate(Task task, [TupleElementNames(new string[] { "this", "observer" })] ValueTuple<TaskObservableExtensions.SlowTaskObservable, IObserver<Unit>> tuple)
					{
						tuple.Item1._scheduler.ScheduleAction(new ValueTuple<Task, IObserver<Unit>>(task, tuple.Item2), delegate([TupleElementNames(new string[] { "task", "observer" })] ValueTuple<Task, IObserver<Unit>> tuple2)
						{
							tuple2.Item1.EmitTaskResult(tuple2.Item2);
						});
					}, new ValueTuple<TaskObservableExtensions.SlowTaskObservable, IObserver<Unit>>(this, observer), cancellationDisposable.Token, taskContinuationOptions);
				}
				return cancellationDisposable;
			}

			private readonly object _task;

			private readonly object _scheduler;
		}

		private sealed class SlowTaskObservable<TResult> : IObservable<TResult>
		{
			public SlowTaskObservable(Task<TResult> task, IScheduler scheduler)
			{
				this._task = task;
				this._scheduler = scheduler;
			}

			public IDisposable Subscribe(IObserver<TResult> observer)
			{
				if (observer == null)
				{
					throw new ArgumentNullException("observer");
				}
				CancellationDisposable cancellationDisposable = new CancellationDisposable();
				TaskContinuationOptions taskContinuationOptions = TaskObservableExtensions.GetTaskContinuationOptions(this._scheduler);
				if (this._scheduler == null)
				{
					this._task.ContinueWith(delegate(Task<TResult> t, object subjectObject)
					{
						t.EmitTaskResult((IObserver<TResult>)subjectObject);
					}, observer, cancellationDisposable.Token, taskContinuationOptions, TaskScheduler.Current);
				}
				else
				{
					this._task.ContinueWithState(delegate(Task<TResult> task, [TupleElementNames(new string[] { "this", "observer" })] ValueTuple<TaskObservableExtensions.SlowTaskObservable<TResult>, IObserver<TResult>> tuple)
					{
						tuple.Item1._scheduler.ScheduleAction(new ValueTuple<Task<TResult>, IObserver<TResult>>(task, tuple.Item2), delegate([TupleElementNames(new string[] { "task", "observer" })] ValueTuple<Task<TResult>, IObserver<TResult>> tuple2)
						{
							tuple2.Item1.EmitTaskResult(tuple2.Item2);
						});
					}, new ValueTuple<TaskObservableExtensions.SlowTaskObservable<TResult>, IObserver<TResult>>(this, observer), cancellationDisposable.Token, taskContinuationOptions);
				}
				return cancellationDisposable;
			}

			private readonly Task<TResult> _task;

			private readonly IScheduler _scheduler;
		}

		private sealed class ToTaskObserver<TResult> : SafeObserver<TResult>
		{
			public ToTaskObserver(TaskCompletionSource<TResult> tcs, CancellationToken ct)
			{
				this._ct = ct;
				this._tcs = tcs;
				if (ct.CanBeCanceled)
				{
					this._ctr = ct.Register(delegate(object @this)
					{
						((TaskObservableExtensions.ToTaskObserver<TResult>)@this).Cancel();
					}, this);
				}
			}

			public override void OnNext(TResult value)
			{
				this._hasValue = true;
				this._lastValue = value;
			}

			public override void OnError(Exception error)
			{
				this._tcs.TrySetException(error);
				this._ctr.Dispose();
				base.Dispose();
			}

			public override void OnCompleted()
			{
				if (this._hasValue)
				{
					this._tcs.TrySetResult(this._lastValue);
				}
				else
				{
					this._tcs.TrySetException(new InvalidOperationException(Strings_Linq.NO_ELEMENTS));
				}
				this._ctr.Dispose();
				base.Dispose();
			}

			private void Cancel()
			{
				base.Dispose();
				this._tcs.TrySetCanceled(this._ct);
			}

			private readonly CancellationToken _ct;

			private readonly TaskCompletionSource<TResult> _tcs;

			private readonly CancellationTokenRegistration _ctr;

			private bool _hasValue;

			private TResult _lastValue;
		}
	}
}
