EtcdRepository implements ConfigurationRepository, RegistryRepository { @Override public void watch(final String key, final DataChangedEventListener dataChangedEventListener) { Watch.Listener listener = Watch.listener(response -> { for (WatchEvent each : response.getEvents()) { ChangedType changedType = getEventChangedType(each); if (ChangedType.IGNORED != changedType) { dataChangedEventListener.onChange(new DataChangedEvent(each.getKeyValue().getKey().toString(Charsets.UTF_8), each.getKeyValue().getValue().toString(Charsets.UTF_8), changedType)); } } }); client.getWatchClient().watch(ByteSequence.from(key, Charsets.UTF_8), listener); }  @Override void init(final String name, final GovernanceCenterConfiguration config); @SneakyThrows({InterruptedException.class, ExecutionException.class}) @Override String get(final String key); @SneakyThrows({InterruptedException.class, ExecutionException.class}) @Override List<String> getChildrenKeys(final String key); @SneakyThrows({InterruptedException.class, ExecutionException.class}) @Override void persist(final String key, final String value); @SneakyThrows({InterruptedException.class, ExecutionException.class}) @Override void persistEphemeral(final String key, final String value); @Override void delete(final String key); @Override void watch(final String key, final DataChangedEventListener dataChangedEventListener); @Override void close(); @Override String getType();  }
@Test public void assertWatchUpdate() { doAnswer(invocationOnMock -> { Watch.Listener listener = (Watch.Listener) invocationOnMock.getArguments()[1]; listener.onNext(buildWatchResponse(WatchEvent.EventType.PUT)); return mock(Watch.Watcher.class); }).when(watch).watch(any(ByteSequence.class), any(Watch.Listener.class)); repository.watch("key1", dataChangedEvent -> { }); verify(watch).watch(any(ByteSequence.class), any(Watch.Listener.class)); }
@Test public void assertWatchDelete() { doAnswer(invocationOnMock -> { Watch.Listener listener = (Watch.Listener) invocationOnMock.getArguments()[1]; listener.onNext(buildWatchResponse(WatchEvent.EventType.DELETE)); return mock(Watch.Watcher.class); }).when(watch).watch(any(ByteSequence.class), any(Watch.Listener.class)); repository.watch("key1", dataChangedEvent -> { }); verify(watch).watch(any(ByteSequence.class), any(Watch.Listener.class)); }
@Test public void assertWatchIgnored() { doAnswer(invocationOnMock -> { Watch.Listener listener = (Watch.Listener) invocationOnMock.getArguments()[1]; listener.onNext(buildWatchResponse(WatchEvent.EventType.UNRECOGNIZED)); return mock(Watch.Watcher.class); }).when(watch).watch(any(ByteSequence.class), any(Watch.Listener.class)); repository.watch("key1", dataChangedEvent -> { }); verify(watch).watch(any(ByteSequence.class), any(Watch.Listener.class)); }