@Test public void testSubscribeToUpdates() throws Exception { zkClient.start(); try { final List<String> addedKeys = new ArrayList<>(); final List<String> removedKeys = new ArrayList<>(); final List<String> changedKeys = new ArrayList<>(); final CountDownLatch added = new CountDownLatch(2); final CountDownLatch removed = new CountDownLatch(1); final CountDownLatch changed = new CountDownLatch(1); DataDictionary.ChangeListener listener = new DataDictionary.ChangeListener() { @Override public synchronized void onKeyAdded(String key) { addedKeys.add(key); added.countDown(); } @Override public synchronized void onKeyRemoved(String key) { removedKeys.add(key); removed.countDown(); } @Override public synchronized void onDataChanged(String key) { changedKeys.add(key); changed.countDown(); } }; DataDictionary d1 = new DataDictionary(zkClient, executor, "foo"); d1.subscribeToUpdates(listener); DataDictionary d2 = new DataDictionary(zkClient, executor, "foo"); d2.write("bar", "bbb".getBytes()); d1.read("bar", listener); d2.write("bar", "aaa".getBytes()); assertThat(changed.await(5, TimeUnit.SECONDS), is(true)); d2.write("baz", "fff".getBytes()); assertThat(added.await(5, TimeUnit.SECONDS), is(true)); zkClient.delete().forPath("/foo/baz"); assertThat(removed.await(5, TimeUnit.SECONDS), is(true)); Collections.sort(addedKeys); assertThat(addedKeys, contains("bar", "baz")); assertThat(removedKeys, contains("baz")); assertThat(changedKeys, contains("bar")); } finally { zkClient.close(); } }
public void subscribeToUpdates(final ChangeListener changeListener) throws Exception { checkNotNull(changeListener); CuratorWatcher watcher = new CuratorWatcher() { @Override public void process(final WatchedEvent event) throws Exception { executor.submit(new Runnable() { @Override public void run() { try { subscribeToUpdates(changeListener); } catch (Exception e) { logger.error("Unable to read all keys from data dictionary", e); throw new RuntimeException(e); } } }); } }; List<String> keys; try { keys = zkClient.getChildren().usingWatcher(watcher).forPath(ZKPaths.makePath(basePath, "")); } catch (KeeperException.NoNodeException e) { createIfNotExists(basePath, null); subscribeToUpdates(changeListener); return; } for (String knownKey : knownKeys) { if (!keys.contains(knownKey)) { if (knownKeys.remove(knownKey)) { changeListener.onKeyRemoved(knownKey); } } } for (String key : keys) { if (knownKeys.add(key)) { changeListener.onKeyAdded(key); } } }
DataDictionary { public void subscribeToUpdates(final ChangeListener changeListener) throws Exception { checkNotNull(changeListener); CuratorWatcher watcher = new CuratorWatcher() { @Override public void process(final WatchedEvent event) throws Exception { executor.submit(new Runnable() { @Override public void run() { try { subscribeToUpdates(changeListener); } catch (Exception e) { logger.error("Unable to read all keys from data dictionary", e); throw new RuntimeException(e); } } }); } }; List<String> keys; try { keys = zkClient.getChildren().usingWatcher(watcher).forPath(ZKPaths.makePath(basePath, "")); } catch (KeeperException.NoNodeException e) { createIfNotExists(basePath, null); subscribeToUpdates(changeListener); return; } for (String knownKey : knownKeys) { if (!keys.contains(knownKey)) { if (knownKeys.remove(knownKey)) { changeListener.onKeyRemoved(knownKey); } } } for (String key : keys) { if (knownKeys.add(key)) { changeListener.onKeyAdded(key); } } } }
DataDictionary { public void subscribeToUpdates(final ChangeListener changeListener) throws Exception { checkNotNull(changeListener); CuratorWatcher watcher = new CuratorWatcher() { @Override public void process(final WatchedEvent event) throws Exception { executor.submit(new Runnable() { @Override public void run() { try { subscribeToUpdates(changeListener); } catch (Exception e) { logger.error("Unable to read all keys from data dictionary", e); throw new RuntimeException(e); } } }); } }; List<String> keys; try { keys = zkClient.getChildren().usingWatcher(watcher).forPath(ZKPaths.makePath(basePath, "")); } catch (KeeperException.NoNodeException e) { createIfNotExists(basePath, null); subscribeToUpdates(changeListener); return; } for (String knownKey : knownKeys) { if (!keys.contains(knownKey)) { if (knownKeys.remove(knownKey)) { changeListener.onKeyRemoved(knownKey); } } } for (String key : keys) { if (knownKeys.add(key)) { changeListener.onKeyAdded(key); } } } @Inject  DataDictionary(CuratorFramework zkClient, ExecutorService executor, @Assisted String basePath); }
DataDictionary { public void subscribeToUpdates(final ChangeListener changeListener) throws Exception { checkNotNull(changeListener); CuratorWatcher watcher = new CuratorWatcher() { @Override public void process(final WatchedEvent event) throws Exception { executor.submit(new Runnable() { @Override public void run() { try { subscribeToUpdates(changeListener); } catch (Exception e) { logger.error("Unable to read all keys from data dictionary", e); throw new RuntimeException(e); } } }); } }; List<String> keys; try { keys = zkClient.getChildren().usingWatcher(watcher).forPath(ZKPaths.makePath(basePath, "")); } catch (KeeperException.NoNodeException e) { createIfNotExists(basePath, null); subscribeToUpdates(changeListener); return; } for (String knownKey : knownKeys) { if (!keys.contains(knownKey)) { if (knownKeys.remove(knownKey)) { changeListener.onKeyRemoved(knownKey); } } } for (String key : keys) { if (knownKeys.add(key)) { changeListener.onKeyAdded(key); } } } @Inject  DataDictionary(CuratorFramework zkClient, ExecutorService executor, @Assisted String basePath); void subscribeToUpdates(final ChangeListener changeListener); int getCurrentVersion(); List<String> getKeys(); void write(String key, byte[] newValue); void write(Map<String, byte[]> newValues); void write(Map<String, byte[]> newValues, int version); void write(List<String> newValues); byte[] read(final String key, final ChangeListener changeListener, boolean notifyWhenAdded); byte[] read(String key, ChangeListener changeListener); byte[] read(String key); }
DataDictionary { public void subscribeToUpdates(final ChangeListener changeListener) throws Exception { checkNotNull(changeListener); CuratorWatcher watcher = new CuratorWatcher() { @Override public void process(final WatchedEvent event) throws Exception { executor.submit(new Runnable() { @Override public void run() { try { subscribeToUpdates(changeListener); } catch (Exception e) { logger.error("Unable to read all keys from data dictionary", e); throw new RuntimeException(e); } } }); } }; List<String> keys; try { keys = zkClient.getChildren().usingWatcher(watcher).forPath(ZKPaths.makePath(basePath, "")); } catch (KeeperException.NoNodeException e) { createIfNotExists(basePath, null); subscribeToUpdates(changeListener); return; } for (String knownKey : knownKeys) { if (!keys.contains(knownKey)) { if (knownKeys.remove(knownKey)) { changeListener.onKeyRemoved(knownKey); } } } for (String key : keys) { if (knownKeys.add(key)) { changeListener.onKeyAdded(key); } } } @Inject  DataDictionary(CuratorFramework zkClient, ExecutorService executor, @Assisted String basePath); void subscribeToUpdates(final ChangeListener changeListener); int getCurrentVersion(); List<String> getKeys(); void write(String key, byte[] newValue); void write(Map<String, byte[]> newValues); void write(Map<String, byte[]> newValues, int version); void write(List<String> newValues); byte[] read(final String key, final ChangeListener changeListener, boolean notifyWhenAdded); byte[] read(String key, ChangeListener changeListener); byte[] read(String key);  }