             [onyx.system :as system :refer [onyx-client]]
             [onyx.extensions :as extensions]
             [onyx.api]
+            [onyx.log.zookeeper :as zk-onyx]
+            [zookeeper :as zk]
             [taoensso.timbre :as timbre :refer [info error spy]]))
 
 (deftemplate page

      (GET  "/" [] (page))
      (GET  "/chsk" req ((:ring-ajax-get-or-ws-handshake sente) req))
      (POST "/chsk" req ((:ring-ajax-post sente) req))
      ;(GET "/cluster/:cluster-id" req (get-job-output req))
      ;(GET "/job/:job-id" req (get-job-output req))
      (resources "/")
      (resources "/react" {:root "react"})
      (route/not-found "Page not found"))

    (launch-event-handler sente peer-config)

    (let [my-ring-handler (ring.middleware.defaults/wrap-defaults my-routes ring-defaults-config)
          server (http-kit-server/run-server my-ring-handler {:port 3000})

   :onyx.peer/job-scheduler :onyx.job-scheduler/round-robin 
   :onyx/id :SUPPLY_ME})

(defn zk-deployment-entry-stat [client entry]
  (:stat (zk/data client (zk-onyx/prefix-path entry))))

(def deployments (atom {}))
(def tracking (atom {}))

(defn distribute-deployments [send-all-fn! deployments]
  (send-all-fn! [:deployment/listing deployments]))

(defn refresh-deployments-watch [send-all-fn! zk-client]
  (->> (zk/children zk-client zk-onyx/root-path :watcher (fn [_] (refresh-deployments-watch send-all-fn! zk-client)))
       (map (juxt identity 
                  (partial zk-deployment-entry-stat zk-client)))
       (map (fn [[child stat]]
              (vector child
                      {:created (java.util.Date. (:ctime stat))
                       :modified (java.util.Date. (:mtime stat))})))
       (into {})
       (reset! deployments)
       (distribute-deployments send-all-fn!)))

(defn event->uid [event]
  (get-in event [:ring-req :cookies "ring-session" :value]))

(defn launch-event-handler [sente peer-config]
  (thread
    (loop []
      (when-let [event (<!! (:ch-chsk sente))]
        (info "EVENT:" event)
        ;; TODO: more sophisticated tracking,
        ;; should track by cluster id rather than user
        ;; and count the number of users tracking it. When the user count drops to 0
        ;; then stop the future.
        (let [user-id (event->uid event)] 
          (case (:id event) 
            :deployment/track (do (when-let [fut (@tracking user-id)]
                                    (future-cancel fut))
                                  (swap! tracking 
                                         assoc 
                                         user-id
                                         (future (track-deployment (:chsk-send! sente)
                                                                   (assoc env-conf :onyx/id (:?data event)) 
                                                                   user-id))))
            :deployment/get-listing ((:chsk-send! sente) user-id [:deployment/listing @deployments])
            :chsk/uidport-close (do
                                  (future-cancel (@tracking user-id))
                                  (swap! tracking dissoc user-id)) 
            (println "Dunno what to do with: " event)))
        (recur)))))

(defn send-mult-fn [send-fn! connected-uids msg]
  (doseq [uid (:any @connected-uids)]
    (send-fn! uid msg)))

; no need for env-config and peer-config
(defrecord HttpServer [peer-config]
  component/Lifecycle
  (start [{:keys [sente] :as component}]
