using System; using UniRx.InternalUtil; namespace UniRx { public sealed class BehaviorSubject : ISubject, IDisposable, IOptimizedObservable { object observerLock = new object(); bool isStopped; bool isDisposed; T lastValue; Exception lastError; IObserver outObserver = EmptyObserver.Instance; public BehaviorSubject(T defaultValue) { lastValue = defaultValue; } public T Value { get { ThrowIfDisposed(); if (lastError != null) lastError.Throw(); return lastValue; } } public bool HasObservers { get { return !(outObserver is EmptyObserver) && !isStopped && !isDisposed; } } public void OnCompleted() { IObserver old; lock (observerLock) { ThrowIfDisposed(); if (isStopped) return; old = outObserver; outObserver = EmptyObserver.Instance; isStopped = true; } old.OnCompleted(); } public void OnError(Exception error) { if (error == null) throw new ArgumentNullException("error"); IObserver old; lock (observerLock) { ThrowIfDisposed(); if (isStopped) return; old = outObserver; outObserver = EmptyObserver.Instance; isStopped = true; lastError = error; } old.OnError(error); } public void OnNext(T value) { IObserver current; lock (observerLock) { if (isStopped) return; lastValue = value; current = outObserver; } current.OnNext(value); } public IDisposable Subscribe(IObserver observer) { if (observer == null) throw new ArgumentNullException("observer"); var ex = default(Exception); var v = default(T); var subscription = default(Subscription); lock (observerLock) { ThrowIfDisposed(); if (!isStopped) { var listObserver = outObserver as ListObserver; if (listObserver != null) { outObserver = listObserver.Add(observer); } else { var current = outObserver; if (current is EmptyObserver) { outObserver = observer; } else { outObserver = new ListObserver(new ImmutableList>(new[] { current, observer })); } } v = lastValue; subscription = new Subscription(this, observer); } else { ex = lastError; } } if (subscription != null) { observer.OnNext(v); return subscription; } else if (ex != null) { observer.OnError(ex); } else { observer.OnCompleted(); } return Disposable.Empty; } public void Dispose() { lock (observerLock) { isDisposed = true; outObserver = DisposedObserver.Instance; lastError = null; lastValue = default(T); } } void ThrowIfDisposed() { if (isDisposed) throw new ObjectDisposedException(""); } public bool IsRequiredSubscribeOnCurrentThread() { return false; } class Subscription : IDisposable { readonly object gate = new object(); BehaviorSubject parent; IObserver unsubscribeTarget; public Subscription(BehaviorSubject parent, IObserver unsubscribeTarget) { this.parent = parent; this.unsubscribeTarget = unsubscribeTarget; } public void Dispose() { lock (gate) { if (parent != null) { lock (parent.observerLock) { var listObserver = parent.outObserver as ListObserver; if (listObserver != null) { parent.outObserver = listObserver.Remove(unsubscribeTarget); } else { parent.outObserver = EmptyObserver.Instance; } unsubscribeTarget = null; parent = null; } } } } } } }