class EtcdWatcherSplittingConsumer implements CheckedBiConsumer<Long, List<WatchEvent>, EtcdException> {

    @Override
    public void accept(Long revision, List<WatchEvent> allWatchEvents) throws EtcdException {
        Builder<ByteSequence, List<WatchEvent>> listsBuilder = builderWithExpectedSize(splitConsumers.size());
        ImmutableSet<ByteSequence> prefixes = splitConsumers.keySet();
        for (ByteSequence keyPrefix : prefixes) {
            listsBuilder.put(keyPrefix, new ArrayList<WatchEvent>());
        }
        ImmutableMap<ByteSequence, List<WatchEvent>> lists = listsBuilder.build();

        for (WatchEvent watchEvent : allWatchEvents) {
            for (ByteSequence keyPrefix : prefixes) {
                if (watchEvent.getKeyValue().getKey().startsWith(keyPrefix)) {
                    lists.get(keyPrefix).add(watchEvent);
                }
            }
        }

        for (Map.Entry<ByteSequence, List<WatchEvent>> list: lists.entrySet()) {
            List<WatchEvent> eventsList = list.getValue();
            // Important optimization, because with separate Oper/Config, one or the other is typically empty
            if (!eventsList.isEmpty()) {
                splitConsumers.get(list.getKey()).accept(eventsList);
            }
        }

        revAwaiter.ifPresent(revAwait -> revAwait.update(revision));
    }

}

class EtcdWatcherSplittingConsumerTest {

    private final TestConsumer configConsumer;
    private final TestConsumer operConsumer;
    private final Map<ByteSequence, CheckedConsumer<List<WatchEvent>, EtcdException>> consumers;

    @Test
    public void testOnlyConfigAndOperAndAnotherOneToIgnore() throws EtcdException {

        EtcdWatcherSplittingConsumer splitter = new EtcdWatcherSplittingConsumer(empty(), consumers);
        splitter.accept(1L, Lists.newArrayList(
                newWatchEvent(append(CONFIGURATION_PREFIX, (byte) 123)),
                newWatchEvent(fromBytes((byte) 234, (byte) 123)),
                newWatchEvent(append(OPERATIONAL_PREFIX, (byte) 123))));
        assertThat(configConsumer.counter.get()).isEqualTo(1L);
        assertThat(operConsumer.counter.get()).isEqualTo(1L);
    }

}