ConsumingRouter to consume based on condition

Question

Started as to make a stream splitter to split stream based on condition, but finally found that I did ConsumingRouter, to consume based on condition and while have to provide consumers before using this router not as splited stream can provide consumers then , and use this router inside forEach of stream, please check it and provide me with your feedback and if there is another way to do it:

public class ConsumingRouter<T> implements Consumer<T> {
    private Map<Predicate<T>, Consumer<T>> map = new ConcurrentHashMap<>();

    public void sub(Predicate<T> p, Consumer<T> c) {
        map.put(p, c);
    }

    @Override
    public void accept(T t) {
        map.entrySet().forEach(e -> {
            if (e.getKey().test(t))
                e.getValue().accept(t);
        });
    }
}

How to use it :

public static void main(String[] args) {
    ConsumingRouter<Integer> router = new ConsumingRouter<Integer>();
    Predicate<Integer> p1 = (i) -> i % 2 != 0;
    Predicate<Integer> p2 = (i) -> i % 2 == 0;
    Consumer<Integer> c1 = i -> {
        System.out.println(Thread.currentThread().getName() + " Route 1 : "
                + i);
    };
    Consumer<Integer> c2 = i -> {
        System.out.println(Thread.currentThread().getName() + " Route 2 : "
                + i);
    };
    router.sub(p1, c1);
    router.sub(p2, c2);
    IntStream.range(0, 1000).boxed().forEach(router);
}

Answer 1

Making the Consumer maintain a mutable map is undesirable. You can make ConsumingRouter an interface with no mutable state:

public interface ConsumingRouter<T> extends Consumer<T> {
    static<T> ConsumingRouter<T> of(Predicate<? super T> p, Consumer<? super T> c) {
        return t -> {if (p.test(t)) {c.accept(t); } };
    }

    default ConsumingRouter<T> andThen(Predicate<? super T> p, Consumer<? super T> c) {
        return this.andThen(ConsumingRouter.of(p, c))::accept;
    }
}

Then in your main you would use it like this:

router = ConsumingRouter.of(p1, c1).andThen(p2,c2);

Now you don't have to worry about somebody adding additional predicates while your stream is computing.

But is it really necessary to have a whole separate class or interface just for this? You could create a simple static method:

public static<T> Consumer<T> acceptIf(Predicate<? super T> p, Consumer<? super T> c) {
    return t -> { if (p.test(t)) { c.accept(t); } };
}

And then use it like this (because Consumer already offers andThen):

router = acceptIf(p1, c1).andThen(acceptIf(p2, c2));

Answer 2

There is a Map.forEach(BiConsumer) that you can use as well:

@Override
public void accept(T t) {
    map.forEach((k,v) -> {
        if (k.test(t)) {
            v.accept(t);
        }
    });
}

Alternatively, you can convert that if-statement in the form of an Optional too if you can ignore null values (as commented by @Misha):

@Override
public void accept(T t) {
    // answer update: switched to Optional.of() for reasons mentioned in comments
    map.forEach((k,v) -> Optional.of(t).filter(k::test).ifPresent(v));
}

You can also take a look at this StackOverflow question for some insight about splitting two Streams into two Streams... similar to yours, except that you are only concerned with multiple Consumers to terminate the source Stream.

edit: Oh yeah, you may want to consider having your sub() (subscribe?) method return the reference to itself, so that you can do method chaining as such:

router.sub(p1, c1).sub(p2, c2);

edit 2: @Misha has a good suggestion for doing away with the Map, if that suits your requirements as well.