(defprotocol ClusterState
 (set-ephemeral-node [this path data])
 (delete-node [this path])
 (create-sequential [this path data])
 ;; if node does not exist, create persistent with this data
 (set-data [this path data])
 (get-data [this path watch?])
 (get-version [this path watch?])
 (get-data-with-version [this path watch?])
 (get-children [this path watch?])
 (mkdirs [this path])
 (close [this])
 (register [this callback])
 (unregister [this id]))

(defprotocol StormClusterState
 (assignments [this callback])
 (assignment-info [this storm-id callback])
 (assignment-info-with-version [this storm-id callback])
 (assignment-version [this storm-id callback])
 (active-storms [this])
 (storm-base [this storm-id callback])
 (get-worker-heartbeat [this storm-id node port])
 (executor-beats [this storm-id executor->node+port])
 (supervisors [this callback])
 (supervisor-info [this supervisor-id]) ;; returns nil if doesn't exist
 (setup-heartbeats! [this storm-id])
 (teardown-heartbeats! [this storm-id])
 (teardown-topology-errors! [this storm-id])
 (heartbeat-storms [this])
 (error-topologies [this])
 (worker-heartbeat! [this storm-id node port info])
 (remove-worker-heartbeat! [this storm-id node port])
 (supervisor-heartbeat! [this supervisor-id info])
 (activate-storm! [this storm-id storm-base])
 (update-storm! [this storm-id new-elems])
 (remove-storm-base! [this storm-id])
 (set-assignment! [this storm-id info])
 (remove-storm! [this storm-id])
 (report-error [this storm-id task-id node port error])
 (errors [this storm-id task-id])
 (disconnect [this]))

(defn mk-distributed-cluster-state
 ;; conf绑定了storm.yaml中的配置信息，是一个map对象
 [conf]
 ;; zk绑定一个zk client，Storm使用CuratorFramework与Zookeeper进行交互
 (let [zk (zk/mk-client conf (conf STORM-ZOOKEEPER-SERVERS) (conf STORM-ZOOKEEPER-PORT) :auth-conf conf)]
   ;; 创建storm集群在zookeeper上的根目录，默认值为/storm
   (zk/mkdirs zk (conf STORM-ZOOKEEPER-ROOT))
   (.close zk))
 ;; callbacks绑定回调函数集合，是一个map对象
 (let [callbacks (atom {})
       ;; active标示zookeeper集群状态
       active (atom true)
       ;; zk重新绑定新的zk client，该zk client设置了watcher，这样当zookeeper集群的状态发生变化时，zk server会给zk client发送相应的event，zk client设置的watcher会调用callbacks中相应回调函数来处理event
       ;; 启动nimbus时，callbacks是一个空集合，所以nimbus端收到event后不会调用任何回调函数；但是启动supervisor时，callbacks中注册了回调函数，所以当supervisor收到zk server发送的event后，会调用相应的回调函数
       ;; mk-client函数定义在zookeeper.clj文件中，请参见其定义部分
   　　zk (zk/mk-client conf
                        (conf STORM-ZOOKEEPER-SERVERS)
                        (conf STORM-ZOOKEEPER-PORT)
                        :auth-conf conf
                        :root (conf STORM-ZOOKEEPER-ROOT)
                        ;; :watcher绑定一个函数，指定zk client的默认watcher函数，state标示当前zk client的状态；type标示事件类型；path标示zookeeper上产生该事件的znode
                        ;; 该watcher函数主要功能就是执行callbacks集合中的函数，callbacks集合中的函数是在mk-storm-cluster-state函数中通过调用ClusterState的register函数添加的
                        :watcher (fn [state type path]
                                   (when @active
                                     (when-not (= :connected state)
                                       (log-warn "Received event " state ":" type ":" path " with disconnected Zookeeper."))
                                     (when-not (= :none type)
                                       (doseq [callback (vals @callbacks)]
                                         (callback type path))))))]
   ;; reify相当于java中的implements，这里表示实现一个协议
   (reify
    ClusterState
    ;; register函数用于将回调函数加入callbacks中，key是一个32位的标识
    (register
      [this callback]
      (let [id (uuid)]
        (swap! callbacks assoc id callback)
        id))
    ;; unregister函数用于将指定key的回调函数从callbacks中删除
    (unregister
      [this id]
      (swap! callbacks dissoc id))
    ;; 在zookeeper上添加一个临时节点
    (set-ephemeral-node
      [this path data]
      (zk/mkdirs zk (parent-path path))
      (if (zk/exists zk path false)
        (try-cause
          (zk/set-data zk path data) ; should verify that it's ephemeral
          (catch KeeperException$NoNodeException e
            (log-warn-error e "Ephemeral node disappeared between checking for existing and setting data")
            (zk/create-node zk path data :ephemeral)
            ))
        (zk/create-node zk path data :ephemeral)))
    ;; 在zookeeper上添加一个顺序节点
    (create-sequential
      [this path data]
      (zk/create-node zk path data :sequential))
    ;; 修改某个节点数据
    (set-data
      [this path data]
      ;; note: this does not turn off any existing watches
      (if (zk/exists zk path false)
        (zk/set-data zk path data)
        (do
          (zk/mkdirs zk (parent-path path))
          (zk/create-node zk path data :persistent))))
    ;; 删除指定节点
    (delete-node
      [this path]
      (zk/delete-recursive zk path))
    ;; 获取指定节点数据。path标示节点路径；watch?是一个布尔类型值，表示是否需要对该节点进行"观察"，如果watch?=true，当调用set-data函数修改该节点数据后，
    ;; 会给zk client发送一个事件，zk client接收事件后，会调用创建zk client时指定的默认watcher函数（即:watcher绑定的函数）
    (get-data
      [this path watch?]
      (zk/get-data zk path watch?))
    ;; 与get-data函数的区别就是获取指定节点数据的同时，获取节点数据的version，version表示节点数据修改的次数
    (get-data-with-version
      [this path watch?]
      (zk/get-data-with-version zk path watch?))
    ;; 获取指定节点的version，watch?的含义与get-data函数中的watch?相同
    (get-version
      [this path watch?]
      (zk/get-version zk path watch?))
    ;; 获取指定节点的子节点列表，watch?的含义与get-data函数中的watch?相同
    (get-children
      [this path watch?]
      (zk/get-children zk path watch?))
    ;; 在zookeeper上创建一个节点
    (mkdirs
      [this path]
      (zk/mkdirs zk path))
    ;; 关闭zk client
    (close
      [this]
      (reset! active false)
      (.close zk)))))

( ))]
         ;; 删除to-kill中包含的节点
         (doseq [k to-kill]
           (delete-node cluster-state (str path "/" k)))))
     ;; 得到给定的storm-id component-id下的异常信息
     (errors
       [this storm-id component-id]
       (let [path (error-path storm-id component-id)
             _ (mkdirs cluster-state path)
             children (get-children cluster-state path false)
             errors (dofor [c children]
                           (let [data (-> (get-data cluster-state (str path "/" c) false)
                                          maybe-deserialize)]
                             (when data
                               (struct TaskError (:error data) (:time-secs data) (:host data) (:port data))
                               )))
             ]
         (->> (filter not-nil? errors)
              (sort-by (comp - :time-secs)))))
     ;; 关闭连接，在关闭连接前，将回调函数从cluster-state的callbacks中删除
     (disconnect
       [this]
       (unregister cluster-state state-id)
       (when solo?
         (close cluster-state))))))

(defnk mk-client
 [conf servers port
  :root ""
  :watcher default-watcher
  :auth-conf nil]
 (let [fk (Utils/newCurator conf servers port root (when auth-conf (ZookeeperAuthInfo. auth-conf)))]
   (.. fk
       (getCuratorListenable)
       (addListener
         (reify CuratorListener
           (^void eventReceived [this ^CuratorFramework _fk ^CuratorEvent e]
                  (when (= (.getType e) CuratorEventType/WATCHED)
                    (let [^WatchedEvent event (.getWatchedEvent e)]
                      (watcher (zk-keeper-states (.getState event))
                               (zk-event-types (.getType event))
                               (.getPath event))))))))
   (.start fk)
   fk))