Rxjava(过滤类)-Filter

只发射通过了谓词测试的数据项

Observable.range(1, 10).filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer % 2 == 0;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});

看一下filter函数

public final Observable<T> filter(Func1<? super T, Boolean> predicate) {
return create(new OnSubscribeFilter<T>(this, predicate));
}

public OnSubscribeFilter(Observable<T> source, Func1<? super T, Boolean> predicate) {
this.source = source;
this.predicate = predicate;
}

predicate就是我们的过滤函数

前面分析可知RangeProducer最终会调用

void fastPath() {
final long endIndex = this.endOfRange + 1L;
final Subscriber<? super Integer> childSubscriber = this.childSubscriber;
for (long index = currentIndex; index != endIndex; index++) {
if (childSubscriber.isUnsubscribed()) {
return;
}
childSubscriber.onNext((int) index);
}
if (!childSubscriber.isUnsubscribed()) {
childSubscriber.onCompleted();
}
}

这里的childSubscriber是FilterSubscriber

public void onNext(T t) {
boolean result;

try {
result = predicate.call(t);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
unsubscribe();
onError(OnErrorThrowable.addValueAsLastCause(ex, t));
return;
}

if (result) {
actual.onNext(t);
} else {
request(1);
}
}

调用过滤函数获取过滤结果，如果为true才调用onNext,达到过滤效果