 ; including chunks that are read e.g. catalog. Then if we end up
 ; with multiple clients viewing the same deployment we can only subscribe once.
 ; Probably not worth the complexity!
-(defn track-deployment [send-fn! client peer-config uid]
-  ; TODO: do we need to close this channel properly when
-  ; cancelling the future so the subscribe-to-log thread doesn't
-  ; keep going?
-  (let [ch (chan 100)]
-    (loop [replica (extensions/subscribe-to-log (:log client) ch)]
-      (let [position (<!! ch)
-            entry (extensions/read-log-entry (:log client) position)
-            new-replica (extensions/apply-log-entry entry replica)
-            diff (extensions/replica-diff entry replica new-replica)
-            job-id (:job (:args entry))
-            tasks (get (:tasks new-replica) job-id)
-            ;complete-tasks (get (:completions new-replica) job-id)
-            ]
-        (case (:fn entry)
-          :submit-job (let [job-id (:id (:args entry))
-                            catalog (extensions/read-chunk (:log client) :catalog job-id)]
-                        (send-fn! uid [:job/submitted-job {:id job-id
-                                                           :entry entry
-                                                           :catalog catalog
-                                                           :created-at (:created-at entry)}]))
-          ; TODO: check if newly submitted jobs will result in peer assigned messages being sent.
-          :volunteer-for-task (let [peer-id (:id (:args entry))
-                                    task (task-allocated-to-peer (:allocations new-replica) peer-id)]  
-                                (if task 
-                                  (send-fn! uid 
-                                            [:job/peer-assigned {:job (:job task)
-                                                                 :task (:task task)
-                                                                 :peer peer-id}])))
-
-          :complete-task (let [task (extensions/read-chunk (:log client) :task (:task (:args entry)))
-                               task-name (:name task)]
-                           (send-fn! uid [:job/completed-task {:name task-name}]))
-          (info "Unable to custom handle entry"))
-        (send-fn! uid [:job/entry entry])
-        ;(send-fn! uid [:deployment/replica {:replica new-replica :diff diff}])
-        (recur new-replica)))))
+(defn track-deployment [send-fn! subscription ch uid]
+  (let [log (:log (:env subscription))]
+    (loop [replica (:replica subscription)]
+      (when-let [position (<!! ch)]
+        (let [entry (extensions/read-log-entry log position)
+              new-replica (extensions/apply-log-entry entry replica)
+              diff (extensions/replica-diff entry replica new-replica)
+              job-id (:job (:args entry))
+              ;complete-tasks (get (:completions new-replica) job-id)
+              tasks (get (:tasks new-replica) job-id)]
+          (case (:fn entry)
+            :submit-job (let [job-id (:id (:args entry))
+                              catalog (extensions/read-chunk log :catalog job-id)]
+                          (send-fn! uid [:job/submitted-job {:id job-id
+                                                             :entry entry
+                                                             :catalog catalog
+                                                             :created-at (:created-at entry)}]))
+            ; TODO: check if newly submitted jobs will result in peer assigned messages being sent.
+            :volunteer-for-task (let [peer-id (:id (:args entry))
+                                      task (task-allocated-to-peer (:allocations new-replica) peer-id)]  
+                                  (if task 
+                                    (send-fn! uid 
+                                              [:job/peer-assigned {:job (:job task)
+                                                                   :task (:task task)
+                                                                   :peer peer-id}])))
+
+            :complete-task (let [task (extensions/read-chunk log :task (:task (:args entry)))
+                                 task-name (:name task)]
+                             (send-fn! uid [:job/completed-task {:name task-name}]))
+            (info "Unable to custom handle entry"))
+          (send-fn! uid [:job/entry entry])
+          ;(send-fn! uid [:deployment/replica {:replica new-replica :diff diff}])
+          (recur new-replica))))))
+
+
+(defrecord LogSubscription [peer-config]
+  component/Lifecycle
+  (start [component]
+    ; Maybe don't need a whole component for this since it already pretty much is one?
+    (info "Start log subscription")
+    (let [sub-ch (chan 100)
+          subscription (onyx.api/subscribe-to-log peer-config sub-ch)] 
+      (assoc component :subscription subscription :subscription-ch sub-ch)))
+  (stop [component]
+    (info "Shutting down log subscription")
+    (onyx.api/shutdown-env (:env (:subscription component)))
+    (assoc component :subscription nil :channel nil)))
+
+
+(defrecord TrackDeploymentManager [send-fn! peer-config user-id]
+  component/Lifecycle
+  (start [component]
+    ; Convert to a system?
+    (info "Starting Track Deployment manager " send-fn! peer-config user-id)
+    (let [subscription (component/start (map->LogSubscription {:peer-config peer-config}))]
+      (assoc component 
+             :subscription subscription
+             :tracking-fut (future 
+                             (track-deployment send-fn!
+                                               (:subscription subscription)
+                                               (:subscription-ch subscription)
+                                               user-id)))))
+  (stop [component]
+    (info "Stopping Track Deployment manager.")
+    (assoc component 
+           :subscription (component/stop (:subscription component))
+           :tracking-fut (future-cancel (:tracking-fut component)))))
+
+(defn new-track-deployment-manager [send-fn! peer-config user-id]
+  (map->TrackDeploymentManager {:send-fn! send-fn! 
+                                :peer-config peer-config 
+                                :user-id user-id}))
+
+(defn stop-tracking! [user-id]
+  (when-let [tracking-manager (@tracking user-id)]
+    (component/stop tracking-manager)
+    (swap! tracking dissoc user-id)))
+
+(defn start-tracking! [send-fn! peer-config deployment-id user-id]
+  (stop-tracking! user-id)
+  (swap! tracking 
+         assoc 
+         user-id
+         (component/start (new-track-deployment-manager send-fn! 
+                                                        (assoc peer-config :onyx/id deployment-id)
+                                                        user-id))))
+
+(defn stop-all-tracking! []
+  (map stop-tracking! (keys @tracking)))
 
 (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 event->uid [event]
  (get-in event [:ring-req :cookies "ring-session" :value]))

; TODO: should be passing tracking around
(defn stop-tracking! [user-id]
  (when-let [current (@tracking user-id)]
    (component/stop (:client current))
    (future-cancel (:tracking-fut current))))

(defn start-tracking! [send-fn! deployment-id peer-config user-id]
  (stop-tracking! user-id)
  (let [client (component/start (system/onyx-client peer-config))] 
    (swap! tracking 
           assoc 
           user-id
           {:client client
            :tracking-fut (future 
                            (track-deployment send-fn!
                                              client
                                              (assoc peer-config :onyx/id deployment-id)
                                              user-id))})))

(defn start-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

            [:security :anti-forgery]
            {:read-token (fn [req] (-> req :params :csrf-token))}))

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

;(deref tracking)

; Improve fn name
(defn job-peer-allocated-task [job-allocations job-id peer-id]
  (ffirst 
