ChildReaper implements Closeable { public void start() throws Exception { Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot be started more than once"); task = executor.scheduleWithFixedDelay ( new Runnable() { @Override public void run() { doWork(); } }, reapingThresholdMs, reapingThresholdMs, TimeUnit.MILLISECONDS ); reaper.start(); } ChildReaper(CuratorFramework client, String path, Reaper.Mode mode);  ChildReaper(CuratorFramework client, String path, Reaper.Mode mode, int reapingThresholdMs);  ChildReaper(CuratorFramework client, String path, Reaper.Mode mode, ScheduledExecutorService executor, int reapingThresholdMs);  ChildReaper(CuratorFramework client, String path, Reaper.Mode mode, ScheduledExecutorService executor, int reapingThresholdMs, String leaderPath); static Set<E> newConcurrentHashSet(); void start(); @Override void close(); ChildReaper addPath(String path); boolean removePath(String path);  }
@Test public void testSomeNodes() throws Exception { Timing timing = new Timing(); ChildReaper reaper = null; CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); try { client.start(); Random r = new Random(); int nonEmptyNodes = 0; for ( int i = 0; i < 10; ++i ) { client.create().creatingParentsIfNeeded().forPath("/test/" + Integer.toString(i)); if ( r.nextBoolean() ) { client.create().forPath("/test/" + Integer.toString(i) + "/foo"); ++nonEmptyNodes; } } reaper = new ChildReaper(client, "/test", Reaper.Mode.REAP_UNTIL_DELETE, 1); reaper.start(); timing.forWaiting().sleepABit(); Stat stat = client.checkExists().forPath("/test"); assertThat(stat.getNumChildren()).isEqualTo(nonEmptyNodes); } finally { CloseableUtils.closeQuietly(reaper); CloseableUtils.closeQuietly(client); } }
@Test public void testSimple() throws Exception { Timing timing = new Timing(); ChildReaper reaper = null; CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); try { client.start(); for ( int i = 0; i < 10; ++i ) { client.create().creatingParentsIfNeeded().forPath("/test/" + Integer.toString(i)); } reaper = new ChildReaper(client, "/test", Reaper.Mode.REAP_UNTIL_DELETE, 1); reaper.start(); timing.forWaiting().sleepABit(); Stat stat = client.checkExists().forPath("/test"); assertThat(stat.getNumChildren()).isZero(); } finally { CloseableUtils.closeQuietly(reaper); CloseableUtils.closeQuietly(client); } }
@Test public void testNamespace() throws Exception { Timing timing = new Timing(); ChildReaper reaper = null; CuratorFramework client = CuratorFrameworkFactory.builder() .connectString(server.getConnectString()) .sessionTimeoutMs(timing.session()) .connectionTimeoutMs(timing.connection()) .retryPolicy(new RetryOneTime(1)) .namespace("foo") .build(); try { client.start(); for ( int i = 0; i < 10; ++i ) { client.create().creatingParentsIfNeeded().forPath("/test/" + Integer.toString(i)); } reaper = new ChildReaper(client, "/test", Reaper.Mode.REAP_UNTIL_DELETE, 1); reaper.start(); timing.forWaiting().sleepABit(); Stat stat = client.checkExists().forPath("/test"); assertThat(stat.getNumChildren()).isZero(); stat = client.usingNamespace(null).checkExists().forPath("/foo/test"); assertThat(stat).isNotNull(); assertThat(stat.getNumChildren()).isZero(); } finally { CloseableUtils.closeQuietly(reaper); CloseableUtils.closeQuietly(client); } }