using System; namespace UniRx.Operators { internal interface ISelect { // IObservable CombineSelector(Func selector); IObservable CombinePredicate(Func predicate); } internal class SelectObservable : OperatorObservableBase, ISelect { readonly IObservable source; readonly Func selector; readonly Func selectorWithIndex; public SelectObservable(IObservable source, Func selector) : base(source.IsRequiredSubscribeOnCurrentThread()) { this.source = source; this.selector = selector; } public SelectObservable(IObservable source, Func selector) : base(source.IsRequiredSubscribeOnCurrentThread()) { this.source = source; this.selectorWithIndex = selector; } // sometimes cause "which no ahead of time (AOT) code was generated." on IL2CPP... //public IObservable CombineSelector(Func combineSelector) //{ // if (this.selector != null) // { // return new Select(source, x => combineSelector(this.selector(x))); // } // else // { // return new Select(this, combineSelector); // } //} public IObservable CombinePredicate(Func predicate) { if (this.selector != null) { return new SelectWhereObservable(this.source, this.selector, predicate); } else { return new WhereObservable(this, predicate); // can't combine } } protected override IDisposable SubscribeCore(IObserver observer, IDisposable cancel) { if (selector != null) { return source.Subscribe(new Select(this, observer, cancel)); } else { return source.Subscribe(new Select_(this, observer, cancel)); } } class Select : OperatorObserverBase { readonly SelectObservable parent; public Select(SelectObservable 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; } observer.OnNext(v); } public override void OnError(Exception error) { try { observer.OnError(error); } finally { Dispose(); } } public override void OnCompleted() { try { observer.OnCompleted(); } finally { Dispose(); } } } // with Index class Select_ : OperatorObserverBase { readonly SelectObservable parent; int index; public Select_(SelectObservable parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { this.parent = parent; this.index = 0; } public override void OnNext(T value) { var v = default(TR); try { v = parent.selectorWithIndex(value, index++); } catch (Exception ex) { try { observer.OnError(ex); } finally { Dispose(); } return; } observer.OnNext(v); } public override void OnError(Exception error) { try { observer.OnError(error); } finally { Dispose(); } } public override void OnCompleted() { try { observer.OnCompleted(); } finally { Dispose(); } } } } }