Add predicate support in the on failure peek operator
@@ -4,19 +4,26 @@ import io.reactivex.Flowable;
import io.smallrye.reactive.Multi;
import java.util.function.Consumer;
+import java.util.function.Predicate;
import static io.smallrye.reactive.helpers.ParameterValidation.nonNull;
public class MultiOnFailurePeek<T> extends MultiOperator<T, T> {
private final Consumer<Throwable> callback;
+ private final Predicate<? super Throwable> predicate;
- public MultiOnFailurePeek(Multi<T> upstream, Consumer<Throwable> callback) {
+ public MultiOnFailurePeek(Multi<T> upstream, Consumer<Throwable> callback, Predicate<? super Throwable> predicate) {
super(nonNull(upstream, "upstream"));
this.callback = nonNull(callback, "callback");
+ this.predicate = predicate == null ? x -> true : predicate;
}
@Override
protected Flowable<T> flowable() {
- return upstreamAsFlowable().doOnError(callback::accept);
+ return upstreamAsFlowable().doOnError(failure -> {
+ if (predicate.test(failure)) {
+ callback.accept(failure);
+ }
+ });
}
}
