SkipWhile<TSource>
namespace System.Reactive.Linq.ObservableImpl
{
internal static class SkipWhile<TSource>
{
internal sealed class Predicate : Producer<TSource, Predicate._>
{
internal sealed class _ : Sink<TSource>, IObserver<TSource>
{
private Func<TSource, bool> _predicate;
public _(Func<TSource, bool> predicate, IObserver<TSource> observer, IDisposable cancel)
: base(observer, cancel)
{
_predicate = predicate;
}
public void OnNext(TSource value)
{
if (_predicate != null) {
bool flag = false;
try {
flag = !_predicate(value);
} catch (Exception error) {
_observer.OnError(error);
base.Dispose();
return;
}
if (flag) {
_predicate = null;
_observer.OnNext(value);
}
} else
_observer.OnNext(value);
}
public void OnError(Exception error)
{
_observer.OnError(error);
base.Dispose();
}
public void OnCompleted()
{
_observer.OnCompleted();
base.Dispose();
}
}
private readonly IObservable<TSource> _source;
private readonly Func<TSource, bool> _predicate;
public Predicate(IObservable<TSource> source, Func<TSource, bool> predicate)
{
_source = source;
_predicate = predicate;
}
protected override _ CreateSink(IObserver<TSource> observer, IDisposable cancel)
{
return new _(_predicate, observer, cancel);
}
protected override IDisposable Run(_ sink)
{
return ObservableExtensions.SubscribeSafe<TSource>(_source, (IObserver<TSource>)sink);
}
}
internal sealed class PredicateIndexed : Producer<TSource, PredicateIndexed._>
{
internal sealed class _ : Sink<TSource>, IObserver<TSource>
{
private Func<TSource, int, bool> _predicate;
private int _index;
public _(Func<TSource, int, bool> predicate, IObserver<TSource> observer, IDisposable cancel)
: base(observer, cancel)
{
_predicate = predicate;
_index = 0;
}
public void OnNext(TSource value)
{
if (_predicate != null) {
bool flag = false;
try {
flag = !_predicate(value, checked(_index++));
} catch (Exception error) {
_observer.OnError(error);
base.Dispose();
return;
}
if (flag) {
_predicate = null;
_observer.OnNext(value);
}
} else
_observer.OnNext(value);
}
public void OnError(Exception error)
{
_observer.OnError(error);
base.Dispose();
}
public void OnCompleted()
{
_observer.OnCompleted();
base.Dispose();
}
}
private readonly IObservable<TSource> _source;
private readonly Func<TSource, int, bool> _predicate;
public PredicateIndexed(IObservable<TSource> source, Func<TSource, int, bool> predicate)
{
_source = source;
_predicate = predicate;
}
protected override _ CreateSink(IObserver<TSource> observer, IDisposable cancel)
{
return new _(_predicate, observer, cancel);
}
protected override IDisposable Run(_ sink)
{
return ObservableExtensions.SubscribeSafe<TSource>(_source, (IObserver<TSource>)sink);
}
}
}
}