using System; namespace UniRx.Operators { internal class SingleObservable : OperatorObservableBase { readonly IObservable source; readonly bool useDefault; readonly Func predicate; public SingleObservable(IObservable source, bool useDefault) : base(source.IsRequiredSubscribeOnCurrentThread()) { this.source = source; this.useDefault = useDefault; } public SingleObservable(IObservable source, Func predicate, bool useDefault) : base(source.IsRequiredSubscribeOnCurrentThread()) { this.source = source; this.predicate = predicate; this.useDefault = useDefault; } protected override IDisposable SubscribeCore(IObserver observer, IDisposable cancel) { if (predicate == null) { return source.Subscribe(new Single(this, observer, cancel)); } else { return source.Subscribe(new Single_(this, observer, cancel)); } } class Single : OperatorObserverBase { readonly SingleObservable parent; bool seenValue; T lastValue; public Single(SingleObservable parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { this.parent = parent; this.seenValue = false; } public override void OnNext(T value) { if (seenValue) { try { observer.OnError(new InvalidOperationException("sequence is not single")); } finally { Dispose(); } } else { seenValue = true; lastValue = value; } } public override void OnError(Exception error) { try { observer.OnError(error); } finally { Dispose(); } } public override void OnCompleted() { if (parent.useDefault) { if (!seenValue) { observer.OnNext(default(T)); } else { observer.OnNext(lastValue); } try { observer.OnCompleted(); } finally { Dispose(); } } else { if (!seenValue) { try { observer.OnError(new InvalidOperationException("sequence is empty")); } finally { Dispose(); } } else { observer.OnNext(lastValue); try { observer.OnCompleted(); } finally { Dispose(); } } } } } class Single_ : OperatorObserverBase { readonly SingleObservable parent; bool seenValue; T lastValue; public Single_(SingleObservable parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { this.parent = parent; this.seenValue = false; } public override void OnNext(T value) { bool isPassed; try { isPassed = parent.predicate(value); } catch (Exception ex) { try { observer.OnError(ex); } finally { Dispose(); } return; } if (isPassed) { if (seenValue) { try { observer.OnError(new InvalidOperationException("sequence is not single")); } finally { Dispose(); } return; } else { seenValue = true; lastValue = value; } } } public override void OnError(Exception error) { try { observer.OnError(error); } finally { Dispose(); } } public override void OnCompleted() { if (parent.useDefault) { if (!seenValue) { observer.OnNext(default(T)); } else { observer.OnNext(lastValue); } try { observer.OnCompleted(); } finally { Dispose(); } } else { if (!seenValue) { try { observer.OnError(new InvalidOperationException("sequence is empty")); } finally { Dispose(); } } else { observer.OnNext(lastValue); try { observer.OnCompleted(); } finally { Dispose(); } } } } } } }