( 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 )))))

( defn mk-storm-cluster-state
? [ cluster-state-spec ]
? ;; satisfies?謂詞相當于java中的instanceof，判斷cluster-state-spec是不是ClusterState實例
?( let [[ solo? cluster-state ] ( if ( satisfies? ClusterState cluster-state-spec )
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? [ false cluster-state-spec ]
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? [ true ( mk-distributed-cluster-state cluster-state-spec )])
? ?　　 ;; 綁定topology id->回調函數的map，當/assignments/{topology id}數據發生變化時，zk client執行assignment-info-callback中topology id所對應的回調函數
? ? ? ? assignment-info-callback ( atom {})
? 　　? ;; assignment-info-with-version-callback與assignment-info-callback類似
? ? ? ? assignment-info-with-version-callback ( atom {})
? ?　　 ;; assignment-version-callback與assignments-callback類似
? ? ? ? assignment-version-callback ( atom {})
? 　　? ;; 當/supervisors標示的znode的子節點發生變化時，zk client執行supervisors-callback指向的函數
? ? ? ? supervisors-callback ( atom nil )
? ?　　 ;; 當/assignments標示的znode的子節點發生變化時，zk client執行assignments-callback指向的函數
? ? ? ? assignments-callback ( atom nil )
? 　　? ;; 當/storms/{topology id}標示的znode的數據發生變化時，zk client執行storm-base-callback中topology id所對應的回調函數
? ? ? ? storm-base-callback ( atom {})
? ?　　 ;; register函數將"回調函數(fn ...)"添加到cluster-state的callbacks集合中，并返回標示該回調函數的uuid
? ? ? ? state-id ( register
? ? ? ? ? ? ? ? ? cluster-state
? ? ? ? ? 　　　　? ;; 定義"回調函數"，type標示事件類型，path標示znode
? ? ? ? ? ? ? ? ? ( fn [ type path ]
? ? ? ? ? ? 　　　　? ;; subtree綁定路徑前綴如"assignments"、"storms"、"supervisors"等，args存放topology id
? ? ? ? ? ? ? ? ? ? ( let [[ subtree & args ] ( tokenize-path path )]
? ? ? ? ? ? ? 　　　　? ;; condp相當于java中的switch
? ? ? ? ? ? ? ? ? ? ? ( condp = subtree
? ? ? ? ? ? ? ? 　　　　? ;; 當subtree="assignments"時，如果args為空，說明是/assignments的子節點發生變化，執行assignments-callback指向的回調函數，否則
? ? ? ? ? ? 　　　　　　? ;; 說明/assignments/{topology id}標示的節點數據發生變化，執行assignment-info-callback指向的回調函數
? ? ? ? ? ? ? ? ? ? ? ? ASSIGNMENTS-ROOT ( if ( empty? args )
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?( issue-callback! assignments-callback )
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?( issue-map-callback! assignment-info-callback ( first args )))
? ? ? ? ? ? ? ?　　　　? ;; 當subtree="supervisors"時，說明是/supervisors的子節點發生變化，執行supervisors-callback指向的回調函數
? ? ? ? ? ? ? ? ? ? ? ? SUPERVISORS-ROOT ( issue-callback! supervisors-callback )
? ? ? ? ? ? 　　　　　　　 ;; 當subtree="storms"時，說明是/storms/{topology id}標示的節點數據發生變化，執行storm-base-callback指向的回調函數
? ? ? ? ? ? ? ? ? ? ? ? STORMS-ROOT ( issue-map-callback! storm-base-callback ( first args ))
? ? ? ? ? ? ? ? ? ? ? ? ;; this should never happen
? ? ? ? ? ? ? ? ? ? ? ? ( exit-process! 30 "Unknown callback for subtree " subtree args )))))]
? ? ;; 在zookeeper上創建storm運行topology所必需的znode
? ?( doseq [ p [ ASSIGNMENTS-SUBTREE STORMS-SUBTREE SUPERVISORS-SUBTREE WORKERBEATS-SUBTREE ERRORS-SUBTREE ]]
? ? ?( mkdirs cluster-state p))
? ? ;; 返回一個實現StormClusterState協議的實例
? ?( reify
? ? ? StormClusterState
? ? ? ;; 獲取/assignments的子節點列表，如果callback不為空，將其賦值給assignments-callback，并對/assignments添加"節點觀察"
? ? ?( assignments
? ? ? ? [ this callback ]
? ? ? ?( when callback
? ? ? ? ?( reset! assignments-callback callback ))
? ? ? ?( get-children cluster-state ASSIGNMENTS-SUBTREE ( not-nil? callback )))
? ? ? ;; 獲取/assignments/{storm-id}節點數據，即storm-id的分配信息，如果callback不為空，將其添加到assignment-info-callback中，并對/assignments/{storm-id}添加"數據觀察"
? ? ?( assignment-info
? ? ? ? [ this storm-id callback ]
? ? ? ?( when callback
? ? ? ? ?( swap! assignment-info-callback assoc storm-id callback ))
? ? ? ?( maybe-deserialize ( get-data cluster-state ( assignment-path storm-id ) ( not-nil? callback ))))
? ? ? ;; 獲取/assignments/{storm-id}節點數據包括version信息，如果callback不為空，將其添加到assignment-info-with-version-callback中，并對/assignments/{storm-id}添加"數據觀察"
? ? ?( assignment-info-with-version
? ? ? ? [ this storm-id callback ]
? ? ? ?( when callback
? ? ? ? ?( swap! assignment-info-with-version-callback assoc storm-id callback ))
? ? ? ?( let [{ data :data version :version }
? ? ? ? ? ? ?( get-data-with-version cluster-state ( assignment-path storm-id ) ( not-nil? callback ))]
? ? ? ? { :data ( maybe-deserialize data )
? ? ? ? :version version }))
? ? ? ;; 獲取/assignments/{storm-id}節點數據的version信息，如果callback不為空，將其添加到assignment-version-callback中，并對/assignments/{storm-id}添加"數據觀察"
? ? ?( assignment-version
? ? ? ? [ this storm-id callback ]
? ? ? ?( when callback
? ? ? ? ?( swap! assignment-version-callback assoc storm-id callback ))
? ? ? ?( get-version cluster-state ( assignment-path storm-id ) ( not-nil? callback )))
? ? ? ;; 獲取storm集群中正在運行的topology id即/storms的子節點列表
? ? ?( active-storms
? ? ? ? [ this ]
? ? ? ?( get-children cluster-state STORMS-SUBTREE false ))
? ? ? ;; 獲取storm集群中所有有心跳的topology id即/workerbeats的子節點列表
? ? ?( heartbeat-storms
? ? ? ? [ this ]
? ? ? ?( get-children cluster-state WORKERBEATS-SUBTREE false ))
? ? ? ;; 獲取所有有錯誤的topology id即/errors的子節點列表
? ? ?( error-topologies
? ? ? ? [ this ]
? ? ? ?( get-children cluster-state ERRORS-SUBTREE false ))
? ? ? ;; 獲取指定storm-id進程的心跳信息，即/workerbeats/{storm-id}/{node-port}節點數據
? ? ?( get-worker-heartbeat
? ? ? ? [ this storm-id node port ]
? ? ? ?( -> cluster-state
? ? ? ? ? ?( get-data ( workerbeat-path storm-id node port ) false )
? ? ? ? ? ? maybe-deserialize ))
? ? ? ;; 獲取指定進程中所有線程的心跳信息
? ? ?( executor-beats
? ? ? ? [ this storm-id executor->node+port ]
? ? ? ? ;; need to take executor->node+port in explicitly so that we don't run into a situation where a
? ? ? ? ;; long dead worker with a skewed clock overrides all the timestamps. By only checking heartbeats
? ? ? ? ;; with an assigned node+port, and only reading executors from that heartbeat that are actually assigned,
? ? ? ? ;; we avoid situations like that
? ? ? ?( let [ node+port->executors ( reverse-map executor->node+port )
? ? ? ? ? ? ? all-heartbeats ( for [[[ node port ] executors ] node+port->executors ]
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ( ->> ( get-worker-heartbeat this storm-id node port )
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?( convert-executor-beats executors )
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ))]
? ? ? ? ?( apply merge all-heartbeats )))
? ? ? ;; 獲取/supervisors的子節點列表，如果callback不為空，將其賦值給supervisors-callback，并對/supervisors添加"節點觀察"
? ? ?( supervisors
? ? ? ? [ this callback ]
? ? ? ?( when callback
? ? ? ? ?( reset! supervisors-callback callback ))
? ? ? ?( get-children cluster-state SUPERVISORS-SUBTREE ( not-nil? callback )))
? ? ? ;; 獲取/supervisors/{supervisor-id}節點數據，即supervisor的心跳信息
? ? ?( supervisor-info
? ? ? ? [ this supervisor-id ]
? ? ? ?( maybe-deserialize ( get-data cluster-state ( supervisor-path supervisor-id ) false )))
? ? ? ;; 設置進程心跳信息
? ? ?( worker-heartbeat!
? ? ? ? [ this storm-id node port info ]
? ? ? ?( set-data cluster-state ( workerbeat-path storm-id node port ) ( Utils/serialize info )))
? ? ? ;; 刪除進程心跳信息
? ? ?( remove-worker-heartbeat!
? ? ? ? [ this storm-id node port ]
? ? ? ?( delete-node cluster-state ( workerbeat-path storm-id node port )))
? ? ? ;; 創建指定storm-id的topology的用于存放心跳信息的節點
? ? ?( setup-heartbeats!
? ? ? ? [ this storm-id ]
? ? ? ?( mkdirs cluster-state ( workerbeat-storm-root storm-id )))
? ? ? ;; 刪除指定storm-id的topology的心跳信息節點
? ? ?( teardown-heartbeats!
? ? ? ? [ this storm-id ]
? ? ? ?( try-cause
? ? ? ? ?( delete-node cluster-state ( workerbeat-storm-root storm-id ))
? ? ? ? ?( catch KeeperException e
? ? ? ? ? ?( log-warn-error e "Could not teardown heartbeats for " storm-id ))))
? ? ? ;; 刪除指定storm-id的topology的錯誤信息節點
? ? ?( teardown-topology-errors!
? ? ? ? [ this storm-id ]
? ? ? ?( try-cause
? ? ? ? ?( delete-node cluster-state ( error-storm-root storm-id ))
? ? ? ? ?( catch KeeperException e
? ? ? ? ? ?( log-warn-error e "Could not teardown errors for " storm-id ))))
? ? ? ;; 創建臨時節點存放supervisor的心跳信息
? ? ?( supervisor-heartbeat!
? ? ? ? [ this supervisor-id info ]
? ? ? ?( set-ephemeral-node cluster-state ( supervisor-path supervisor-id ) ( Utils/serialize info )))
? ? ? ;; 創建/storms/{storm-id}節點
? ? ?( activate-storm!
? ? ? ? [ this storm-id storm-base ]
? ? ? ?( set-data cluster-state ( storm-path storm-id ) ( Utils/serialize storm-base )))
? ? ? ;; 更新topology對應的StormBase對象，即更新/storm/{storm-id}節點
? ? ?( update-storm!
? ? ? ? [ this storm-id new-elems ]
? ? ? ? ;; base綁定storm-id在zookeeper上的StormBase對象
? ? ? ?( let [ base ( storm-base this storm-id nil )
? ? ? ? ? ? ? ;; executors綁定component名稱->組件并行度的map
? ? ? ? ? ? ? executors ( :component->executors base )
? ? ? ? ? ? ? ;; new-elems綁定合并后的組件并行度map，update函數將組件新并行度map合并到舊map中
? ? ? ? ? ? ? new-elems ( update new-elems :component->executors ( partial merge executors ))]
? ? ? ? ? ;; 更新StormBase對象中的組件并行度map，并寫入zookeeper的/storms/{storm-id}節點
? ? ? ? ?( set-data cluster-state ( storm-path storm-id )
? ? ? ? ? ? ? ? ? ?( -> base
? ? ? ? ? ? ? ? ? ? ? ?( merge new-elems )
? ? ? ? ? ? ? ? ? ? ? ? Utils/serialize ))))
? ? ? ;; 獲取storm-id的StormBase對象，即讀取/storms/{storm-id}節點數據，如果callback不為空，將其賦值給storm-base-callback，并為/storms/{storm-id}節點添加"數據觀察"
? ? ?( storm-base
? ? ? ? [ this storm-id callback ]
? ? ? ?( when callback
? ? ? ? ?( swap! storm-base-callback assoc storm-id callback ))
? ? ? ?( maybe-deserialize ( get-data cluster-state ( storm-path storm-id ) ( not-nil? callback ))))
? ? ? ;; 刪除storm-id的StormBase對象，即刪除/storms/{storm-id}節點
? ? ?( remove-storm-base!
? ? ? ? [ this storm-id ]
? ? ? ?( delete-node cluster-state ( storm-path storm-id )))
? ? ? ;; 更新storm-id的分配信息，即更新/assignments/{storm-id}節點數據
? ? ?( set-assignment!
? ? ? ? [ this storm-id info ]
? ? ? ?( set-data cluster-state ( assignment-path storm-id ) ( Utils/serialize info )))
? ? ? ;; 刪除storm-id的分配信息，同時刪除其StormBase信息，即刪除/assignments/{storm-id}節點和/storms/{storm-id}節點
? ? ?( remove-storm!
? ? ? ? [ this storm-id ]
? ? ? ?( delete-node cluster-state ( assignment-path storm-id ))
? ? ? ?( remove-storm-base! this storm-id ))
? ? ? ;; 將組件異常信息寫入zookeeper
? ? ?( report-error
? ? ? ? [ this storm-id component-id node port error ]
? ? ? ? ;; path綁定"/errors/{storm-id}/{component-id}"
? ? ? ?( let [ path ( error-path storm-id component-id )
? ? ? ? ? ? ? ;; data綁定異常信息，包括異常時間、異常堆棧信息、主機和端口
? ? ? ? ? ? ? data { :time-secs ( current-time-secs ) :error ( stringify-error error ) :host node :port port }
? ? ? ? ? ? ? ;; 創建/errors/{storm-id}/{component-id}節點
? ? ? ? ? ? ? _ ( mkdirs cluster-state path )
? ? ? ? ? ? ? ;; 創建/errors/{storm-id}/{component-id}的子順序節點，并寫入異常信息
? ? ? ? ? ? ? _ ( create-sequential cluster-state ( str path "/e" ) ( Utils/serialize data ))
? ? ? ? ? ? ? ;; to-kill綁定除去順序節點編號最大的前10個節點的剩余節點的集合
? ? ? ? ? ? ? to-kill ( ->> ( get-children cluster-state path false )
? ? ? ? ? ? ? ? ? ? ? ? ? ( sort-by parse-error-path )
? ? ? ? ? ? ? ? ? ? ? ? ? reverse
? ? ? ? ? ? ? ? ? ? ? ? ? ( drop 10 ))]
? ? ? ? ? ;; 刪除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 ))