Routing Java objects using conditional consumers

Question

As per a previous question here and answers I got, I used it to get a new implementation. I found that I have to make two implementations match once "to ignore other consumers after matching once" and match all "to execute all consumers its filter matching the input", and made a common interface to abstract usage of two implementations:

Common interface

public interface Router<T> {
Router<T> chain(Predicate<T> p, Consumer<T> c);
Consumer<T> toConsumer();
}

Match All:

public interface RouterMatchAll<T> extends Consumer<T>, Router<T> {
static <T> RouterMatchAll<T> of(Predicate<T> p1, Consumer<T> c1) {
    BiFunction<Predicate<T>, Consumer<T>, RouterMatchAll<T>> preFun = (p, c) -> i -> Optional
            .ofNullable(i).filter(p).ifPresent(c);
    return preFun.apply(p1, c1);
}

@Override
public default RouterMatchAll<T> chain(Predicate<T> p, Consumer<T> c) {
    return this.andThen(RouterMatchAll.of(p, c))::accept;
}

@Override
public default Consumer<T> toConsumer() {
    return this;
}

}

Match Once:

public interface RouterMatchOnce<T> extends Function<Optional<T>, Optional<T>>,
    Router<T> {

static <T> RouterMatchOnce<T> of(Predicate<T> p1, Consumer<T> c1) {
    BiFunction<Predicate<T>, Consumer<T>, RouterMatchOnce<T>> preFun = (p,
            c) -> o -> {
        if (o.filter(p).isPresent()) {
            c.accept(o.get());
            return Optional.empty();
        }
        return o;
    };
    return preFun.apply(p1, c1);

}

@Override
public default RouterMatchOnce<T> chain(Predicate<T> p, Consumer<T> c) {
    return this.andThen(RouterMatchOnce.of(p, c))::apply;
}

@Override
public default Consumer<T> toConsumer() {
    return i -> this.apply(Optional.ofNullable(i));
}

}

For MatchAll, I used Consumer to use its andThen, which consumes the same input. For MatchOnce, I had to use Function<Optional,Optional> and its andThen, use the output of first function to be used in next function, and I needed to change the data to Optional.empty() after consuming once to not consume it then.

Concerns I want your feedback about:

1. toConsumer() in RouterMatchAll is just return this and I don't like it but couldn't find another option for abstraction.
2. Names are not nice as it is not a router any more. It has to be something like filter, interceptor, or ConditionalConsumer. Please suggest a name for these interfaces, chain, and toConsumer methods.

Usage of the above:

Predicate<Integer> p1 = (Integer i) -> {
    System.out.println("call p1");
    return i < 5;
};
Predicate<Integer> p2 = (Integer i) -> {
    System.out.println("call p2");
    return i > 5;
};
Consumer<Integer> c1 = i -> System.out.println("less");
Consumer<Integer> c2 = i -> System.out.println("more");

Router<Integer> routerMatchAll = RouterMatchAll.of(p1, c1)
        .chain(p2, c2);
IntStream.range(0, 10).boxed().forEach(routerMatchAll.toConsumer());
System.out.println(">>>>>>>>>>>>>>>>>>>>>>");
Router<Integer> routerMatchOnce = RouterMatchOnce.of(p1, c1).chain(p2,
        c2);
IntStream.range(0, 10).boxed().forEach(routerMatchOnce.toConsumer());

Output:

  call p1
  less
  call p2
  call p1
  less
  call p2
  call p1
  less
  call p2
  call p1
  less
  call p2
  call p1
  less
  call p2
  call p1
  call p2
  call p1
  call p2
  more
  call p1
  call p2
  more
  call p1
  call p2
  more
  call p1
  call p2
  more
  >>>>>>>>>>>>>>>>>>>>>>
  call p1
  less
  call p1
  less
  call p1
  less
  call p1
  less
  call p1
  less
  call p1
  call p2
  call p1
  call p2
  more
  call p1
  call p2
  more
  call p1
  call p2
  more
  call p1
  call p2
  more

Answer 1

Again, I strongly recommend you stick to the methods and systems already provided in the Streams API. In this case, the ability to observe-without-changing the stream is given with the Stream.peek() method and the ability to conditionally remove a value from the stream is given with the Stream.filter() method. Using these two methods with a simple helper class, you can get the functionality you desire. Consider your existing "use case":

Predicate<Integer> p1 = (Integer i) -> {
    System.out.println("call p1");
    return i < 5;
};
Predicate<Integer> p2 = (Integer i) -> {
    System.out.println("call p2");
    return i > 5;
};
Consumer<Integer> c1 = i -> System.out.println("less");
Consumer<Integer> c2 = i -> System.out.println("more");

Router<Integer> routerMatchAll = RouterMatchAll.of(p1, c1)
        .chain(p2, c2);
IntStream.range(0, 10).boxed().forEach(routerMatchAll.toConsumer());
System.out.println(">>>>>>>>>>>>>>>>>>>>>>");
Router<Integer> routerMatchOnce = RouterMatchOnce.of(p1, c1).chain(p2,
        c2);
IntStream.range(0, 10).boxed().forEach(routerMatchOnce.toConsumer());

How about a system where you would implement the first stream:

Router<Integer> routerMatchAll = RouterMatchAll.of(p1, c1)
        .chain(p2, c2);
IntStream.range(0, 10).boxed().forEach(routerMatchAll.toConsumer());

as:

    IntStream.range(0, 10).boxed()
        .peek(new ConditionalTee<>(p1, c1))
        .forEach(new ConditionalTee<>(p2, c2));

Note how you use peek() to ensure all values pass through to the forEach()?

The second implementation is currently:

Router<Integer> routerMatchOnce = RouterMatchOnce.of(p1, c1).chain(p2,
        c2);
IntStream.range(0, 10).boxed().forEach(routerMatchOnce.toConsumer());

but I would instead suggest:

    IntStream.range(0, 10).boxed()
        .filter(new ConditionalTee<>(p1, c1))
        .forEach(new ConditionalTee<>(p2, c2));

Note how we call filter() instead of peek()?

Now, what is ConditionalTee? It is a simple class that takes a predicate, and a Consumer, and it calls the consumer with the value, if the value passes, and it returns false if the predicate passes (which means the value will be filtered in a filter call). It is more comments than code, and it's really simple:

public class ConditionalTee<T> implements Predicate<T>, Consumer<T> {

    private final Predicate<? super T> predicate;
    private final Consumer<? super T> consumer;

    public ConditionalTee(
          Predicate<? super T> condition, Consumer<? super T> consumer) {
        this.predicate = condition;
        this.consumer = consumer;
    }

    @Override
    public void accept(T t) {
        test(t);
    }

    @Override
    public boolean test(T t) {
        // NOTE: inverted boolean.
        // this will be called from a Stream.filter()
        // need to return true if the value should stay on the stream
        // false if it should be removed from the stream.
        // We want to keep it on the stream if we are not siphoning it off
        //     (return true).
        // We want to remove from the stream if we are siphoning the value off
        //     (return false).
        if (!predicate.test(t)) {
            return true;
        }
        consumer.accept(t);
        return false;
    }
}

See it all running here, in ideone.