WatcherRemovalManager { void removeWatchers() { if ( client.isZk34CompatibilityMode() ) { return; } List<NamespaceWatcher> localEntries = Lists.newArrayList(entries); while ( localEntries.size() > 0 ) { NamespaceWatcher watcher = localEntries.remove(0); if ( entries.remove(watcher) && !client.isZk34CompatibilityMode() ) { try { log.debug("Removing watcher for path: " + watcher.getUnfixedPath()); RemoveWatchesBuilderImpl builder = new RemoveWatchesBuilderImpl(client); builder.internalRemoval(watcher, watcher.getUnfixedPath()); } catch ( Exception e ) { log.error("Could not remove watcher for path: " + watcher.getUnfixedPath()); } } } } WatcherRemovalManager(CuratorFrameworkImpl client);   }
@Test public void testSameWatcherDifferentKinds() throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)); try { client.start(); WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework(); Watcher watcher = new Watcher() { @Override public void process(WatchedEvent event) { } }; removerClient.create().creatingParentsIfNeeded().forPath("/a/b/c"); removerClient.checkExists().usingWatcher(watcher).forPath("/a/b/c"); removerClient.getData().usingWatcher(watcher).forPath("/a/b/c"); removerClient.removeWatchers(); } finally { TestCleanState.closeAndTestClean(client); } }
@Test public void testMissingNode() throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)); try { client.start(); WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework(); Watcher w = new Watcher() { @Override public void process(WatchedEvent event) { } }; try { removerClient.getData().usingWatcher(w).forPath("/one/two/three"); Assert.fail("Should have thrown NoNodeException"); } catch ( KeeperException.NoNodeException expected ) { } removerClient.removeWatchers(); } finally { TestCleanState.closeAndTestClean(client); } }
@Test public void testSameWatcherDifferentPaths1Triggered() throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)); try { client.start(); WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework(); final CountDownLatch latch = new CountDownLatch(1); Watcher watcher = new Watcher() { @Override public void process(WatchedEvent event) { latch.countDown(); } }; removerClient.checkExists().usingWatcher(watcher).forPath("/a/b/c"); removerClient.checkExists().usingWatcher(watcher).forPath("/d/e/f"); removerClient.create().creatingParentsIfNeeded().forPath("/d/e/f"); Timing timing = new Timing(); Assert.assertTrue(timing.awaitLatch(latch)); timing.sleepABit(); removerClient.removeWatchers(); } finally { TestCleanState.closeAndTestClean(client); } }
@Test public void testSameWatcherDifferentKinds1Triggered() throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)); try { client.start(); WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework(); final CountDownLatch latch = new CountDownLatch(1); Watcher watcher = new Watcher() { @Override public void process(WatchedEvent event) { latch.countDown(); } }; removerClient.create().creatingParentsIfNeeded().forPath("/a/b/c"); removerClient.checkExists().usingWatcher(watcher).forPath("/a/b/c"); removerClient.getData().usingWatcher(watcher).forPath("/a/b/c"); removerClient.setData().forPath("/a/b/c", "new".getBytes()); Timing timing = new Timing(); Assert.assertTrue(timing.awaitLatch(latch)); timing.sleepABit(); removerClient.removeWatchers(); } finally { TestCleanState.closeAndTestClean(client); } }