@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); }
@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 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 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); }
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); }
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);  }