DataDictionary { public void write(String key, byte[] newValue) throws Exception { Map<String, byte[]> entry = new HashMap<>(); entry.put(key, newValue); write(entry); } @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);  }
@Test public void testDataChangeWatcherMultipleWatches() throws Exception { zkClient.start(); try { DataDictionary d1 = new DataDictionary(zkClient, executor, "foo"); DataDictionary d2 = new DataDictionary(zkClient, executor, "foo"); d2.write("bar", "fff".getBytes()); Semaphore sync = new Semaphore(1); sync.acquire(); DataWatcherExample dataWatcher = new DataWatcherExample(d1, sync); assertThat(dataWatcher.readAndWatchKey("bar"), is("fff".getBytes())); assertThat(sync.tryAcquire(5, TimeUnit.SECONDS), is(true)); d2.write("bar", "aaa".getBytes()); assertThat(sync.tryAcquire(5, TimeUnit.SECONDS), is(true)); d2.write("bar", "bbb".getBytes()); assertThat(sync.tryAcquire(5, TimeUnit.SECONDS), is(true)); assertThat(dataWatcher.getDataHistory().size(), is(3)); assertThat(dataWatcher.getDataHistory(), hasItems("fff".getBytes(), "aaa".getBytes(), "bbb".getBytes())); } finally { zkClient.close(); } }