using System;
using System.Linq.Expressions;
using System.Reactive.Disposables;
using System.Reflection;
using System.Runtime.Remoting;
using System.Runtime.Remoting.Lifetime;
using System.Security;
using System.Threading;

namespace System.Reactive.Linq
{
	public static class RemotingObservable
	{
		public static IObservable<TSource> Remotable<TSource>(this IObservable<TSource> source)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			return RemotingObservable.Remotable_<TSource>(source);
		}

		public static IObservable<TSource> Remotable<TSource>(this IObservable<TSource> source, ILease lease)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			return RemotingObservable.Remotable_<TSource>(source, lease);
		}

		public static IQbservable<TSource> Remotable<TSource>(this IQbservable<TSource> source)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			return source.Provider.CreateQuery<TSource>(Expression.Call(null, ((MethodInfo)MethodBase.GetCurrentMethod()).MakeGenericMethod(new Type[] { typeof(TSource) }), new Expression[] { source.Expression }));
		}

		public static IQbservable<TSource> Remotable<TSource>(this IQbservable<TSource> source, ILease lease)
		{
			if (source == null)
			{
				throw new ArgumentNullException("source");
			}
			return source.Provider.CreateQuery<TSource>(Expression.Call(null, ((MethodInfo)MethodBase.GetCurrentMethod()).MakeGenericMethod(new Type[] { typeof(TSource) }), source.Expression, Expression.Constant(lease, typeof(ILease))));
		}

		private static IObservable<TSource> Remotable_<TSource>(IObservable<TSource> source)
		{
			return new RemotingObservable.SerializableObservable<TSource>(new RemotingObservable.RemotableObservable<TSource>(source, null));
		}

		private static IObservable<TSource> Remotable_<TSource>(IObservable<TSource> source, object lease)
		{
			return new RemotingObservable.SerializableObservable<TSource>(new RemotingObservable.RemotableObservable<TSource>(source, lease));
		}

		[Serializable]
		private class SerializableObservable<T> : IObservable<T>
		{
			public SerializableObservable(RemotingObservable.RemotableObservable<T> remotableObservable)
			{
				this._remotableObservable = remotableObservable;
			}

			public IDisposable Subscribe(IObserver<T> observer)
			{
				ISafeObserver<T> safeObserver = SafeObserver<T>.Wrap(observer);
				IDisposable disposable = this._remotableObservable.Subscribe(new RemotingObservable.RemotableObserver<T>(safeObserver));
				safeObserver.SetResource(disposable);
				return disposable;
			}

			private readonly RemotingObservable.RemotableObservable<T> _remotableObservable;
		}

		private class RemotableObserver<T> : MarshalByRefObject, IObserver<T>, ISponsor
		{
			public RemotableObserver(IObserver<T> underlyingObserver)
			{
				this._underlyingObserver = underlyingObserver;
			}

			public void OnNext(T value)
			{
				this._underlyingObserver.OnNext(value);
			}

			public void OnError(Exception exception)
			{
				try
				{
					this._underlyingObserver.OnError(exception);
				}
				finally
				{
					this.Unregister();
				}
			}

			public void OnCompleted()
			{
				try
				{
					this._underlyingObserver.OnCompleted();
				}
				finally
				{
					this.Unregister();
				}
			}

			[SecuritySafeCritical]
			private void Unregister()
			{
				ILease lease = (ILease)RemotingServices.GetLifetimeService(this);
				if (lease != null)
				{
					lease.Unregister(this);
				}
			}

			[SecurityCritical]
			public override object InitializeLifetimeService()
			{
				ILease lease = (ILease)base.InitializeLifetimeService();
				lease.Register(this);
				return lease;
			}

			[SecurityCritical]
			TimeSpan ISponsor.Renewal(ILease lease)
			{
				return lease.InitialLeaseTime;
			}

			private readonly IObserver<T> _underlyingObserver;
		}

		[Serializable]
		private sealed class RemotableObservable<T> : MarshalByRefObject, IObservable<T>
		{
			public RemotableObservable(IObservable<T> underlyingObservable, ILease lease)
			{
				this._underlyingObservable = underlyingObservable;
				this._lease = lease;
			}

			public IDisposable Subscribe(IObserver<T> observer)
			{
				return new RemotingObservable.RemotableObservable<T>.RemotableSubscription(this._underlyingObservable.Subscribe(observer));
			}

			[SecurityCritical]
			public override object InitializeLifetimeService()
			{
				return this._lease;
			}

			private readonly IObservable<T> _underlyingObservable;

			private readonly ILease _lease;

			private sealed class RemotableSubscription : MarshalByRefObject, IDisposable, ISponsor
			{
				public RemotableSubscription(IDisposable underlyingSubscription)
				{
					this._underlyingSubscription = underlyingSubscription;
				}

				public void Dispose()
				{
					using (Interlocked.Exchange<IDisposable>(ref this._underlyingSubscription, Disposable.Empty))
					{
						this.Unregister();
					}
				}

				[SecuritySafeCritical]
				private void Unregister()
				{
					ILease lease = (ILease)RemotingServices.GetLifetimeService(this);
					if (lease != null)
					{
						lease.Unregister(this);
					}
				}

				[SecurityCritical]
				public override object InitializeLifetimeService()
				{
					ILease lease = (ILease)base.InitializeLifetimeService();
					lease.Register(this);
					return lease;
				}

				[SecurityCritical]
				TimeSpan ISponsor.Renewal(ILease lease)
				{
					return lease.InitialLeaseTime;
				}

				private IDisposable _underlyingSubscription;
			}
		}
	}
}
