# Routing Java objects using conditional consumers

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.

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

• Actually, do you mind sharing how else will these implementations be used? As mentioned in my answer to your earlier question, since you only care about consuming a single stream differently, a decidedly simpler solution is to use something like Collections.partitioningBy() and then call forEach() on the resulting Map's values... Sep 23 '15 at 0:22
• My problem with Collectors.partitioningBy or Collectors.groupBy that it is expecting cold stream "finite stream" to be collected at end of operation, but what if my stream is still running and I don't want to collect it yet, check another question I have about infinitestream below how to use collectors with it and it is still running codereview.stackexchange.com/questions/105287/… Sep 23 '15 at 2:05
• BTW in your earlier question as well, @Misha pointed out one pitfall with my initial Optional-based suggestion regarding null-handling. If that affects your implementation, then you should switch to your original if-statement too. :) Sep 23 '15 at 2:28
• :) your and @Misha answers derived me to this implementation, and if you noticed I had to use if statement in RouterMatchOnce as I have to consume based of false to, on stackoverlow I have OptionalConsumer Answer but I don't want to use it here to not mix things, her the link stackoverflow.com/questions/23773024/… may I'll add it as question here too. Sep 23 '15 at 2:34

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


    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.

• ConditionalTee is returning true if "!predicate.test(t)" and calling "consumer.accept(t);" inside test method it is confusing,peek is not going to make routing, could be used for duplicate objects to be consumed by 2 consumers and make condition on consumer, and for conditional consumer will only use i->Optional.of(i).filter(p).ifPresent(c); no need for this class to do it. Sep 23 '15 at 13:03
• @BassemRedaZohdy - your comment is really hard to understand. I feel you have a broken understanding of how Streams are supposed to be used. In each of your recent questions you are doing things which are "backwards" in your implementation, and I don't know how to say that more politely, but there was a time when streams were an "upstream" fight for me to use, but then it "clicked" and I now use the streams to help with the flow. My impression is that you are still trying to swim upstream, and you will have more success if you use the streams in the directions they naturally work. Sep 23 '15 at 13:24
• let me describe to you my original problem with streams, streams is perfect for me but only have 1-2 issues with it, and this Q is way to solve one of them, which was I want to do something for condition and something else for another condition, lets represents it as if-statment and switch case, filter for me as if-then, and there no else so how to represents else, and how is stream to represents switch case, there are Collectors partitioningBy and groupingBy but it will work only with cold stream not hot (infinite) stream, so How to solve this? this is the Q Sep 23 '15 at 13:34