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(); if (!eventsList.isEmpty()) { splitConsumers.get(list.getKey()).accept(eventsList); } } revAwaiter.ifPresent(revAwait -> revAwait.update(revision)); } EtcdWatcherSplittingConsumer(Optional<RevAwaiter> revAwaiter,
            Map<ByteSequence, CheckedConsumer<List<WatchEvent>, EtcdException>> splitConsumers); @Override void accept(Long revision, List<WatchEvent> allWatchEvents);  }
@Test public void testEmpty() throws EtcdException { EtcdWatcherSplittingConsumer splitter = new EtcdWatcherSplittingConsumer(empty(), consumers); splitter.accept(1L, Collections.emptyList()); assertThat(configConsumer.counter.get()).isEqualTo(0L); assertThat(operConsumer.counter.get()).isEqualTo(0L); }
@Test public void testOnlyConfig() throws EtcdException { EtcdWatcherSplittingConsumer splitter = new EtcdWatcherSplittingConsumer(empty(), consumers); splitter.accept(1L, Lists.newArrayList(newWatchEvent(append(CONFIGURATION_PREFIX, (byte)123)))); assertThat(configConsumer.counter.get()).isEqualTo(1L); assertThat(operConsumer.counter.get()).isEqualTo(0L); }
@Test public void testOnlyOper() throws EtcdException { EtcdWatcherSplittingConsumer splitter = new EtcdWatcherSplittingConsumer(empty(), consumers); splitter.accept(1L, Lists.newArrayList(newWatchEvent(append(OPERATIONAL_PREFIX, (byte)123)))); assertThat(configConsumer.counter.get()).isEqualTo(0L); assertThat(operConsumer.counter.get()).isEqualTo(1L); }
@Test public void testOnlyConfigAndOper() throws EtcdException { EtcdWatcherSplittingConsumer splitter = new EtcdWatcherSplittingConsumer(empty(), consumers); splitter.accept(1L, Lists.newArrayList( newWatchEvent(append(CONFIGURATION_PREFIX, (byte) 123)), newWatchEvent(append(OPERATIONAL_PREFIX, (byte) 123)))); assertThat(configConsumer.counter.get()).isEqualTo(1L); assertThat(operConsumer.counter.get()).isEqualTo(1L); }
@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); }