using System; namespace UniRx.Operators { // Optimize for .Select().Where() internal class SelectWhereObservable : OperatorObservableBase { readonly IObservable source; readonly Func selector; readonly Func predicate; public SelectWhereObservable(IObservable source, Func selector, Func predicate) : base(source.IsRequiredSubscribeOnCurrentThread()) { this.source = source; this.selector = selector; this.predicate = predicate; } protected override IDisposable SubscribeCore(IObserver observer, IDisposable cancel) { return source.Subscribe(new SelectWhere(this, observer, cancel)); } class SelectWhere : OperatorObserverBase { readonly SelectWhereObservable parent; public SelectWhere(SelectWhereObservable parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { this.parent = parent; } public override void OnNext(T value) { var v = default(TR); try { v = parent.selector(value); } catch (Exception ex) { try { observer.OnError(ex); } finally { Dispose(); } return; } var isPassed = false; try { isPassed = parent.predicate(v); } catch (Exception ex) { try { observer.OnError(ex); } finally { Dispose(); } return; } if (isPassed) { observer.OnNext(v); } } public override void OnError(Exception error) { try { observer.OnError(error); } finally { Dispose(); } } public override void OnCompleted() { try { observer.OnCompleted(); } finally { Dispose(); } } } } }