fix bug: blocking on close

Author: vran-dev

src/main/java/cc/cc1234/main/PrettyZooApplication.java

@@ -1,5 +1,7 @@
 package cc.cc1234.main;
 
+import cc.cc1234.main.cache.ZkClientCache;
+import cc.cc1234.main.cache.ZkListenerCache;
 import cc.cc1234.main.controller.NodeTreeViewController;
 import javafx.application.Application;
 import javafx.fxml.FXMLLoader;
@@ -32,6 +34,14 @@ public static void showNodeTreeView(Stage primary) throws IOException {
         NodeTreeViewController controller = loader.getController();
         controller.setPrimaryStage(primary);
         primary.show();
+
+    }
+
+    @Override
+    public void stop() throws Exception {
+        super.stop();
+        ZkListenerCache.getInstance().closeAll();
+        ZkClientCache.getInstance().closeAll();
     }
 
     public static void main(String[] args) {

src/main/java/cc/cc1234/main/cache/ZkClientCache.java

@@ -27,4 +27,10 @@ public CuratorFramework put(String server, CuratorFramework client) {
         return CACHE.put(server, client);
     }
 
+    public void closeAll() {
+        CACHE.forEach((key, client) -> {
+            client.close();
+        });
+    }
+
 }

src/main/java/cc/cc1234/main/cache/ZkListenerCache.java

@@ -0,0 +1,33 @@
+package cc.cc1234.main.cache;
+
+import cc.cc1234.main.manager.CuratorlistenerManager;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ZkListenerCache {
+
+    private static final Map<String, CuratorlistenerManager> CACHE = new ConcurrentHashMap<>();
+
+    private static final ZkListenerCache ZK_LISTENER_CACHE = new ZkListenerCache();
+
+    public static ZkListenerCache getInstance() {
+        return ZK_LISTENER_CACHE;
+    }
+
+    public void put(String server, CuratorlistenerManager manager) {
+        CACHE.put(server, manager);
+    }
+
+    public CuratorlistenerManager get(String server) {
+        return CACHE.get(server);
+    }
+
+    public static boolean contains(String server) {
+        return CACHE.containsKey(server);
+    }
+
+    public void closeAll() {
+        CACHE.forEach((key, manager) -> manager.close());
+    }
+}

src/main/java/cc/cc1234/main/controller/NodeTreeViewController.java

@@ -2,6 +2,7 @@
 
 import cc.cc1234.main.cache.TreeItemCache;
 import cc.cc1234.main.cache.ZkClientCache;
+import cc.cc1234.main.cache.ZkListenerCache;
 import cc.cc1234.main.curator.NodeEventHandler;
 import cc.cc1234.main.history.History;
 import cc.cc1234.main.manager.CuratorlistenerManager;
@@ -19,14 +20,15 @@
 import org.apache.curator.framework.recipes.cache.TreeCacheListener;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class NodeTreeViewController {
 
+    private static final Logger LOG = LoggerFactory.getLogger(NodeTreeViewController.class);
+
     @FXML
     private TreeView<ZkNode> zkNodeTreeView;
 
@@ -81,10 +83,7 @@ public class NodeTreeViewController {
 
     private ZkClientCache serverClientCache = ZkClientCache.getInstance();
 
-    /**
-     * [server:manager]
-     */
-    private Map<String, CuratorlistenerManager> curatorlistenerManagerCache = new ConcurrentHashMap<>();
+    public static ZkListenerCache zkListenerCache = ZkListenerCache.getInstance();
 
     /**
      * current selected zk server
@@ -98,6 +97,9 @@ public class NodeTreeViewController {
 
     public void setPrimaryStage(Stage primary) {
         this.primaryStage = primary;
+        primary.onCloseRequestProperty().addListener((observable, oldValue, newValue) -> {
+            serverClientCache.closeAll();
+        });
     }
 
     @FXML
@@ -185,10 +187,11 @@ private CuratorFramework doConnect(ZkServer server) {
             // TODO @vran use connection listener
             client.blockUntilConnected();
         } catch (InterruptedException e) {
-            throw new IllegalStateException(e);
+            VToast.toastFailure(primaryStage, "Connect zookeeper failed");
+            LOG.error("", e);
         }
 
-        Executors.newSingleThreadExecutor().execute(() -> {
+        Platform.runLater(() -> {
             final String value = history.get(server.getServer(), "0");
             history.save(server.getServer(), String.valueOf(Integer.parseInt(value) + 1));
             history.store();
@@ -201,9 +204,10 @@ void syncTreeNode(String server, CuratorFramework client) {
     }
 
     private void startSyncTreeNodeListener(String server, CuratorFramework client) {
-        CuratorlistenerManager manager = curatorlistenerManagerCache.getOrDefault(server,
-                new CuratorlistenerManager(client));
-        manager.start(new TreeCacheListener() {
+        if (!zkListenerCache.contains(server)) {
+            zkListenerCache.put(server, new CuratorlistenerManager(client));
+        }
+        zkListenerCache.get(server).start(new TreeCacheListener() {
 
             private NodeEventHandler eventHandler = new NodeEventHandler(server);
 
@@ -245,8 +249,8 @@ private void updateDataAction() {
                     .inBackground((client, event) -> Platform.runLater(() -> VToast.toastSuccess(primaryStage)))
                     .forPath(this.pathLabel.getText(), this.dataTextArea.getText().getBytes());
         } catch (Exception e) {
-            VToast.toastFailure(primaryStage);
-            throw new IllegalStateException(e);
+            VToast.toastFailure(primaryStage, "update data failed");
+            LOG.error("update data error", e);
         }
     }
 

src/main/java/cc/cc1234/main/manager/CuratorlistenerManager.java

@@ -24,4 +24,8 @@ public void start(TreeCacheListener listener) {
         }
     }
 
+    public void close() {
+        treeCache.close();
+    }
+
 }