JEtcdClient extends AbstractEtcdClient<JEtcdClient.EtcdWatcher> { @Override public void delete(String path) { clientWrapper.delete(path); } JEtcdClient(URL url); @Override void doCreatePersistent(String path); @Override long doCreateEphemeral(String path); @Override boolean checkExists(String path); @Override EtcdWatcher createChildWatcherListener(String path, ChildListener listener); @Override List<String> addChildWatcherListener(String path, EtcdWatcher etcdWatcher); @Override void removeChildWatcherListener(String path, EtcdWatcher etcdWatcher); @Override List<String> getChildren(String path); @Override boolean isConnected(); @Override long createLease(long second); @Override long createLease(long ttl, long timeout, TimeUnit unit); @Override void delete(String path); @Override void revokeLease(long lease); @Override void doClose(); @Override String getKVValue(String key); @Override boolean put(String key, String value); @Override boolean putEphemeral(String key, String value); ManagedChannel getChannel();  }
@Test public void test_watch_when_create_path() throws InterruptedException { String path = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers"; String child = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers/demoService1"; final CountDownLatch notNotified = new CountDownLatch(1); ChildListener childListener = (parent, children) -> { Assertions.assertEquals(1, children.size()); Assertions.assertEquals(child.substring(child.lastIndexOf("/") + 1), children.get(0)); notNotified.countDown(); }; client.addChildListener(path, childListener); client.createEphemeral(child); Assertions.assertTrue(notNotified.await(10, TimeUnit.SECONDS)); client.removeChildListener(path, childListener); client.delete(child); }
@Test public void test_watch_when_create_wrong_path() throws InterruptedException { String path = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers"; String child = "/dubbo/com.alibaba.dubbo.demo.DemoService/routers/demoService1"; final CountDownLatch notNotified = new CountDownLatch(1); ChildListener childListener = (parent, children) -> { Assertions.assertEquals(1, children.size()); Assertions.assertEquals(child, children.get(0)); notNotified.countDown(); }; client.addChildListener(path, childListener); client.createEphemeral(child); Assertions.assertFalse(notNotified.await(1, TimeUnit.SECONDS)); client.removeChildListener(path, childListener); client.delete(child); }
@Test public void test_watch_when_delete_path() throws InterruptedException { String path = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers"; String child = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers/demoService1"; final CountDownLatch notNotified = new CountDownLatch(1); ChildListener childListener = (parent, children) -> { Assertions.assertEquals(0, children.size()); notNotified.countDown(); }; client.createEphemeral(child); client.addChildListener(path, childListener); client.delete(child); Assertions.assertTrue(notNotified.await(10, TimeUnit.SECONDS)); client.removeChildListener(path, childListener); }
@Test public void test_watch_then_unwatch() throws InterruptedException { String path = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers"; String child = "/dubbo/com.alibaba.dubbo.demo.DemoService/providers/demoService2"; final CountDownLatch notNotified = new CountDownLatch(1); final CountDownLatch notTwiceNotified = new CountDownLatch(2); final Holder notified = new Holder(); ChildListener childListener = (parent, children) -> { Assertions.assertEquals(1, children.size()); Assertions.assertEquals(child.substring(child.lastIndexOf("/") + 1), children.get(0)); notNotified.countDown(); notTwiceNotified.countDown(); notified.getAndIncrease(); }; client.addChildListener(path, childListener); client.createEphemeral(child); Assertions.assertTrue(notNotified.await(15, TimeUnit.SECONDS)); client.removeChildListener(path, childListener); client.delete(child); Assertions.assertFalse(notTwiceNotified.await(5, TimeUnit.SECONDS)); Assertions.assertEquals(1, notified.value); client.delete(child); }
@Test public void test_watch_on_recoverable_connection() throws InterruptedException { String path = "/dubbo/com.alibaba.dubbo.demo.DemoService/connection"; String child = "/dubbo/com.alibaba.dubbo.demo.DemoService/connection/demoService1"; final CountDownLatch notNotified = new CountDownLatch(1); final CountDownLatch notTwiceNotified = new CountDownLatch(2); final Holder notified = new Holder(); ChildListener childListener = (parent, children) -> { notTwiceNotified.countDown(); switch (notified.increaseAndGet()) { case 1: { notNotified.countDown(); Assertions.assertEquals(1, children.size()); Assertions.assertEquals(child.substring(child.lastIndexOf("/") + 1), children.get(0)); break; } case 2: { Assertions.assertEquals(0, children.size()); Assertions.assertEquals(path, parent); break; } default: Assertions.fail("two many callback invoked."); } }; client.addChildListener(path, childListener); client.createEphemeral(child); Assertions.assertTrue(notNotified.await(15, TimeUnit.SECONDS)); JEtcdClient.EtcdWatcher watcher = client.getChildListener(path, childListener); watcher.onError(Status.UNAVAILABLE.withDescription("temporary connection issue").asRuntimeException()); client.delete(child); Assertions.assertTrue(notTwiceNotified.await(15, TimeUnit.SECONDS)); client.removeChildListener(path, childListener); }