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(); }
}
}
}
}