diff --git a/src/main/java/com/alibaba/druid/pool/ha/node/ZookeeperNodeListener.java b/src/main/java/com/alibaba/druid/pool/ha/node/ZookeeperNodeListener.java
new file mode 100644
index 0000000000..37b223b243
--- /dev/null
+++ b/src/main/java/com/alibaba/druid/pool/ha/node/ZookeeperNodeListener.java
@@ -0,0 +1,317 @@
+/*
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.druid.pool.ha.node;
+
+import com.alibaba.druid.DruidRuntimeException;
+import com.alibaba.druid.pool.ha.PropertiesUtils;
+import com.alibaba.druid.support.logging.Log;
+import com.alibaba.druid.support.logging.LogFactory;
+import com.alibaba.druid.util.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A NodeListner that watches a Zookeeper Path.
+ * e.g.
+ * Path to watch:
+ *
+ * + ha-druid-datasources
+ * |---- NodeA
+ * |---- NodeB
+ *
+ *
+ * The Data of NodeA (in Java properties format):
+ *
+ * foo.host=xxx.xxx.xxx.xxx
+ * foo.port=3308
+ * foo.username=foo
+ * foo.password=foo_password
+ *
+ * bar.host=xxx.xxx.xxx.xxx
+ * bar.port=3309
+ * bar.username=bar
+ * bar.password=bar_password
+ *
+ *
+ * @author DigitalSonic
+ */
+public class ZookeeperNodeListener extends NodeListener {
+ private final static Log LOG = LogFactory.getLog(ZookeeperNodeListener.class);
+ private String zkConnectString;
+ private String path = "/ha-druid-datasources";
+ private String dataPrefix = "";
+ private Lock lock = new ReentrantLock();
+ private boolean privateZkClient = false; // Should I close the client?
+ private PathChildrenCache cache;
+ private CuratorFramework client;
+ /**
+ * Template Format:
+ * jdbc:mysql://${host}:${port}/foo?useUnicode=true
+ * ${host} and ${port} will be replaced by values in ZK
+ */
+ private String urlTemplate;
+
+ /**
+ * Init a PathChildrenCache to watch the given path.
+ */
+ @Override
+ public void init() {
+ checkParameters();
+ super.init();
+ if (client == null) {
+ client = CuratorFrameworkFactory.builder()
+ .canBeReadOnly(true)
+ .connectionTimeoutMs(3000)
+ .connectString(zkConnectString)
+ .retryPolicy(new ExponentialBackoffRetry(5000, 3, 30000))
+ .sessionTimeoutMs(30000)
+ .build();
+ client.start();
+ privateZkClient = true;
+ }
+ cache = new PathChildrenCache(client, path, true);
+ cache.getListenable().addListener(new PathChildrenCacheListener() {
+ @Override
+ public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
+ try {
+ lock.lock();
+ LOG.info("Receive an event: " + event.getType());
+ PathChildrenCacheEvent.Type eventType = event.getType();
+ switch (eventType) {
+ case CHILD_REMOVED:
+ updateSingleNode(event, NodeEventTypeEnum.DELETE);
+ break;
+ case CHILD_ADDED:
+ updateSingleNode(event, NodeEventTypeEnum.ADD);
+ break;
+ case CONNECTION_RECONNECTED:
+ refreshAllNodes();
+ break;
+ default:
+ // CHILD_UPDATED
+ // INITIALIZED
+ // CONNECTION_LOST
+ // CONNECTION_SUSPENDED
+ LOG.info("Received a PathChildrenCacheEvent, IGNORE it: " + event);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ });
+ try {
+ // Use BUILD_INITIAL_CACHE to force build cache in the current Thread.
+ // We don't use POST_INITIALIZED_EVENT, so there's no INITIALIZED event.
+ cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
+ } catch (Exception e) {
+ LOG.error("Can't start PathChildrenCache", e);
+ }
+ }
+
+ /**
+ * Close PathChildrenCache and CuratorFramework.
+ */
+ @Override
+ public void destroy() {
+ if (cache != null) {
+ try {
+ cache.close();
+ } catch (IOException e) {
+ LOG.error("IOException occurred while closing PathChildrenCache.", e);
+ }
+ }
+ if (client != null && privateZkClient) {
+ client.close();
+ }
+ }
+
+ /**
+ * Build Properties from PathChildrenCache.
+ * Should be called after init().
+ *
+ * @see #getPropertiesFromCache()
+ */
+ @Override
+ public List refresh() {
+ Properties properties = getPropertiesFromCache();
+ List events = NodeEvent.getEventsByDiffProperties(getProperties(), properties);
+ if (events != null && !events.isEmpty()) {
+ setProperties(properties);
+ }
+ return events;
+ }
+
+ private void checkParameters() {
+ if (client == null && StringUtils.isEmpty(zkConnectString)) {
+ throw new DruidRuntimeException("ZK Client is NULL, Please set the zkConnectString.");
+ }
+ if (StringUtils.isEmpty(path)) {
+ throw new DruidRuntimeException("Please set the ZooKeeper node path.");
+ }
+ if (StringUtils.isEmpty(urlTemplate)) {
+ throw new DruidRuntimeException("Please set the urlTemplate.");
+ }
+ }
+
+ private void updateSingleNode(PathChildrenCacheEvent event, NodeEventTypeEnum type) {
+ ChildData data = event.getData();
+ String nodeName = getNodeName(data);
+ List names = new ArrayList();
+ names.add(dataPrefix + "." + nodeName);
+ Properties properties = getPropertiesFromChildData(data);
+ List events = NodeEvent.generateEvents(properties, names, type);
+
+ if (events.isEmpty()) {
+ return;
+ }
+ if (type == NodeEventTypeEnum.ADD) {
+ getProperties().putAll(properties);
+ } else {
+ for (String n : properties.stringPropertyNames()) {
+ getProperties().remove(n);
+ }
+ }
+ super.update(events);
+ }
+
+ private void refreshAllNodes() {
+ try {
+ if (client.checkExists().forPath(path) == null) {
+ LOG.warn("PATH[" + path + "] is NOT existed, can NOT refresh nodes.");
+ return;
+ }
+ cache.rebuild();
+ Properties properties = getPropertiesFromCache();
+ List events = NodeEvent.getEventsByDiffProperties(getProperties(), properties);
+ if (events != null && !events.isEmpty()) {
+ setProperties(properties);
+ super.update(events);
+ }
+ } catch (Exception e) {
+ LOG.error("Can NOT refresh Cache Nodes.", e);
+ }
+ }
+
+ private Properties getPropertiesFromCache() {
+ List data = cache.getCurrentData();
+ Properties properties = new Properties();
+ for (ChildData c : data) {
+ properties.putAll(getPropertiesFromChildData(c));
+ }
+ return properties;
+ }
+
+ private Properties getPropertiesFromChildData(ChildData data) {
+ Properties properties = new Properties();
+ if (data == null) {
+ return properties;
+ }
+ String nodeName = getNodeName(data);
+ String str = new String(data.getData());
+ Properties full = new Properties();
+ try {
+ full.load(new StringReader(str));
+ } catch (IOException e) {
+ LOG.error("Can't load Properties from String. " + str, e);
+ }
+ Properties filtered = PropertiesUtils.filterPrefix(full, dataPrefix);
+ for (String n : filtered.stringPropertyNames()) {
+ properties.setProperty(
+ n.replaceFirst(dataPrefix, dataPrefix + "." + nodeName),
+ filtered.getProperty(n));
+ }
+ if (!properties.containsKey(dataPrefix + "." + nodeName + ".url")) {
+ properties.setProperty(dataPrefix + "." + nodeName + ".url", formatUrl(filtered));
+ }
+ return properties;
+ }
+
+ private String formatUrl(Properties properties) {
+ String url = urlTemplate;
+ if (properties.containsKey(dataPrefix + ".host")) {
+ url = url.replace("${host}", properties.getProperty(dataPrefix + ".host"));
+ }
+ if (properties.containsKey(dataPrefix + ".port")) {
+ url = url.replace("${port}", properties.getProperty(dataPrefix + ".port"));
+ }
+ return url;
+ }
+
+ private String getNodeName(ChildData data) {
+ String eventZkPath = data.getPath();
+ if (eventZkPath.startsWith(path + "/")) {
+ return eventZkPath.substring(path.length() + 1);
+ } else {
+ return eventZkPath;
+ }
+ }
+
+ public void setClient(CuratorFramework client) {
+ if (client != null) {
+ this.client = client;
+ privateZkClient = false;
+ }
+ }
+
+ public CuratorFramework getClient() {
+ return client;
+ }
+
+ public String getZkConnectString() {
+ return zkConnectString;
+ }
+
+ public void setZkConnectString(String zkConnectString) {
+ this.zkConnectString = zkConnectString;
+ }
+
+ public String getPath() {
+ return path;
+ }
+
+ public void setPath(String path) {
+ this.path = path;
+ }
+
+ public String getDataPrefix() {
+ return dataPrefix;
+ }
+
+ public void setDataPrefix(String dataPrefix) {
+ this.dataPrefix = dataPrefix;
+ }
+
+ public String getUrlTemplate() {
+ return urlTemplate;
+ }
+
+ public void setUrlTemplate(String urlTemplate) {
+ this.urlTemplate = urlTemplate;
+ }
+}
diff --git a/src/test/java/com/alibaba/druid/pool/ha/node/ZookeeperNodeListenerTest.java b/src/test/java/com/alibaba/druid/pool/ha/node/ZookeeperNodeListenerTest.java
new file mode 100644
index 0000000000..8eedda1167
--- /dev/null
+++ b/src/test/java/com/alibaba/druid/pool/ha/node/ZookeeperNodeListenerTest.java
@@ -0,0 +1,163 @@
+package com.alibaba.druid.pool.ha.node;
+
+import com.alibaba.druid.DruidRuntimeException;
+import com.alibaba.druid.support.logging.Log;
+import com.alibaba.druid.support.logging.LogFactory;
+import org.apache.curator.test.TestingServer;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Observable;
+import java.util.Observer;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ZookeeperNodeListenerTest {
+ private final static Log LOG = LogFactory.getLog(ZookeeperNodeListenerTest.class);
+ private static TestingServer server;
+ private final String PATH = "/ha-druid-datasource";
+
+ @BeforeClass
+ public static void init() throws Exception {
+ server = new TestingServer();
+ }
+
+ @AfterClass
+ public static void destroy() throws Exception {
+ server.close();
+ }
+
+ @Test
+ public void testAddOneNode() throws Exception {
+ CountDownLatch cdl = new CountDownLatch(1);
+ ZookeeperNodeListener listener = new ZookeeperNodeListener();
+ listener.setZkConnectString(server.getConnectString());
+ listener.setPath(PATH);
+ listener.setUrlTemplate("jdbc:mysql://${host}:${port}/foo");
+ listener.setDataPrefix("foo");
+ listener.setObserver(new Observer() {
+ @Override
+ public void update(Observable o, Object arg) {
+ NodeEvent[] events = (NodeEvent[]) arg;
+ assertEquals(1, events.length);
+ NodeEvent event = events[0];
+ LOG.info("NodeEvent received: " + event);
+ if (NodeEventTypeEnum.ADD == event.getType()) {
+ assertEquals("foo", event.getUsername());
+ assertEquals("password", event.getPassword());
+ assertEquals("foo.test-foo", event.getNodeName());
+ assertEquals("jdbc:mysql://127.0.0.1:1234/foo",
+ event.getUrl());
+ cdl.countDown();
+ }
+ }
+ });
+ listener.init();
+
+ ZookeeperNodeRegister register = registerNodeAndReturnRegister();
+ Thread.sleep(2000); // Wait for the Node to be created.
+
+ cdl.await(10, TimeUnit.SECONDS);
+ Properties properties = listener.getProperties();
+ assertEquals(5, properties.size());
+ assertEquals("foo", properties.getProperty("foo.test-foo.username"));
+ assertEquals("password", properties.getProperty("foo.test-foo.password"));
+ assertEquals("127.0.0.1", properties.getProperty("foo.test-foo.host"));
+ assertEquals("1234", properties.getProperty("foo.test-foo.port"));
+ assertEquals("jdbc:mysql://127.0.0.1:1234/foo",
+ properties.getProperty("foo.test-foo.url"));
+ register.destroy();
+ listener.destroy();
+ }
+
+ @Test
+ public void testRemoveOneNode() throws Exception {
+ ZookeeperNodeRegister register = registerNodeAndReturnRegister();
+ Thread.sleep(2000); // Wait for the Node to be created.
+
+ CountDownLatch addCDL = new CountDownLatch(1);
+ CountDownLatch removeCDL = new CountDownLatch(1);
+ ZookeeperNodeListener listener = new ZookeeperNodeListener();
+ listener.setZkConnectString(server.getConnectString());
+ listener.setPath(PATH);
+ listener.setUrlTemplate("jdbc:mysql://${host}:${port}/foo");
+ listener.setDataPrefix("foo");
+ listener.setObserver(new Observer() {
+ @Override
+ public void update(Observable o, Object arg) {
+ NodeEvent[] events = (NodeEvent[]) arg;
+ assertEquals(1, events.length);
+ NodeEvent event = events[0];
+ LOG.info("NodeEvent received: " + event);
+ if (NodeEventTypeEnum.DELETE == event.getType()) {
+ assertEquals("foo", event.getUsername());
+ assertEquals("password", event.getPassword());
+ assertEquals("foo.test-foo", event.getNodeName());
+ assertEquals("jdbc:mysql://127.0.0.1:1234/foo",
+ event.getUrl());
+ removeCDL.countDown();
+ } else {
+ addCDL.countDown();
+ }
+ }
+ });
+ listener.init();
+ listener.update();
+ addCDL.await(10, TimeUnit.SECONDS);
+ Properties properties = listener.getProperties();
+ assertEquals(5, properties.size());
+
+ register.destroy();
+ removeCDL.await(10, TimeUnit.SECONDS);
+ assertTrue(properties.isEmpty());
+ listener.destroy();
+ }
+
+ private ZookeeperNodeRegister registerNodeAndReturnRegister() {
+ ZookeeperNodeRegister register = new ZookeeperNodeRegister();
+ register.setZkConnectString(server.getConnectString());
+ register.setPath(PATH);
+ register.init();
+
+ List payload = new ArrayList();
+ ZookeeperNodeInfo node = new ZookeeperNodeInfo();
+ node.setPrefix("foo");
+ node.setHost("127.0.0.1");
+ node.setPort(1234);
+ node.setUsername("foo");
+ node.setPassword("password");
+ payload.add(node);
+ register.register("test-foo", payload);
+ return register;
+ }
+
+ @Test
+ public void testInitFailure() {
+ ZookeeperNodeListener listener = new ZookeeperNodeListener();
+ try {
+ listener.init();
+ fail("zkClient is null and zkConnectString is null");
+ } catch (DruidRuntimeException e) {
+ }
+ listener.setZkConnectString("127.0.0.1:2181");
+ try {
+ listener.init();
+ fail("path is null");
+ } catch (DruidRuntimeException e) {
+ }
+ listener.setPath(PATH);
+ try {
+ listener.init();
+ fail("urlTemplate is null");
+ } catch (DruidRuntimeException e) {
+ }
+ }
+}