 (defn storm-active? [storm-cluster-state storm-name]
   (not-nil? (get-storm-id storm-cluster-state storm-name)))
 
+(defn check-storm-active! [nimbus storm-name active?]
+  (if (= (not active?)
+         (storm-active? (:storm-cluster-state nimbus)
+                        storm-name))
+    (if active?
+      (throw (NotAliveException. (str storm-name " is not alive")))
+      (throw (AlreadyAliveException. (str storm-name " is already active"))))
+    ))
+
 (defn topologies-to-kill [storm-cluster-state]
   (let [assigned-ids (set (.assignments storm-cluster-state nil))]
     (reduce (fn [ret id]

    ;; TODO: validate that every declared stream is not a system stream
    ))

(defn file-cache-map [conf]
  (TimeCacheMap.
   (int (conf NIMBUS-FILE-COPY-EXPIRATION-SECS))
   (reify TimeCacheMap$ExpiredCallback
          (expire [this id stream]
                  (.close stream)
                  ))
   ))

(defn extract-status-str [base]
  (let [t (-> base :status :type)]
    (.toUpperCase (name t))

    (assoc storm-conf TOPOLOGY-KRYO-REGISTER sers)
    ))

(defn set-topology-status! [nimbus storm-name status]
  (let [storm-cluster-state (:storm-cluster-state nimbus)        
        storm-id (get-storm-id storm-cluster-state storm-name)]
   (when-not storm-id
     (throw (NotAliveException. storm-name)))
   (locking (:submit-lock nimbus)
     (.update-storm! storm-cluster-state
                     storm-id
                     {:status status}))
   (log-message "Updated " storm-name " with status " status)
   ))

(defn mk-cleanup-fn [nimbus]
  (fn []
    (let [storm-cluster-state (:storm-cluster-state nimbus)
          conf (:conf nimbus)
          submit-lock (:submit-lock nimbus)
          id->kill-time (locking submit-lock
                          (topologies-to-kill storm-cluster-state))]
      (when-not (empty? id->kill-time)
        (log-message "Entering kill loop for " (pr-str id->kill-time)))
      (loop [kill-order (sort-by second (seq id->kill-time))]
        (when-not (empty? kill-order)
          (let [[id kill-time] (first kill-order)
                now (current-time-secs)]
            (if (or (nil? kill-time) (>= now kill-time))
              (do
                ;; technically a supervisor could still think there's an assignment and try to d/l
                ;; this will cause supervisor to go down and come back up... eventually it should sync
                ;; TODO: removing code locally should be done separately (since topology that doesn't start will still have code)
                (log-message "Killing topology: " id)
                (rmr (master-stormdist-root conf id))
                (.remove-storm! storm-cluster-state id)
                (recur (rest kill-order))
                )
              (do
                (log-message "Waiting for " (- kill-time now) " seconds to kill topology " id)
                (sleep-secs (- kill-time now))
                (recur kill-order)
                ))
            )))
      (when-not (empty? id->kill-time)
        (log-message "Killed " (pr-str id->kill-time)))
      (let [to-cleanup-ids (locking submit-lock
                             (cleanup-storm-ids conf storm-cluster-state))]
        (when-not (empty? to-cleanup-ids)
          (doseq [id to-cleanup-ids]
            (.teardown-heartbeats! storm-cluster-state id)
            (.teardown-task-errors! storm-cluster-state id)
            (swap! (:task-heartbeats-cache nimbus) dissoc id)
            )
          (log-message "Cleaned up topology task heartbeats: " (pr-str to-cleanup-ids))
          )))))

(defn mk-reassign-fn [nimbus]
  (fn this []
    (let [conf (:conf nimbus)
          submit-lock (:submit-lock nimbus)
          storm-cluster-state (:storm-cluster-state nimbus)]
      (when (conf NIMBUS-REASSIGN)
        (locking submit-lock
          (let [callback (fn [& ignored] (.add (:event-manager nimbus) this))
                active-storm-ids (.active-storms storm-cluster-state)]
            (doseq [storm-id active-storm-ids]
              (let [base (.storm-base storm-cluster-state storm-id nil)]
                (if (= :active (-> base :status :type))
                  (mk-assignments conf
                                  storm-id
                                  storm-cluster-state
                                  callback
                                  (:task-heartbeats-cache nimbus)))))
            ))))))

(defserverfn service-handler [conf]
  (let [nimbus (nimbus-data conf)        
        cleanup-fn (mk-cleanup-fn nimbus)
        reassign-fn (mk-reassign-fn nimbus)
        threads [(async-loop
                  (fn []
                    (.add (:event-manager nimbus) reassign-fn)
                    (.add (:cleanup-manager nimbus) cleanup-fn)
                    (when @(:active nimbus) (conf NIMBUS-MONITOR-FREQ-SECS))
                    ))
                 ]]
    (reify Nimbus$Iface
      (^void submitTopology
        [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology]
        (check-storm-active! nimbus storm-name false)
        (validate-topology! topology)
        (swap! (:submitted-count nimbus) inc)
        (let [storm-id (str storm-name "-" @(:submitted-count nimbus) "-" (current-time-secs))
              storm-conf (normalize-conf
                          conf
                          (-> serializedConf
                              from-json
                              (assoc STORM-ID storm-id)
                              ))
              total-storm-conf (merge conf storm-conf)
              topology (if (total-storm-conf TOPOLOGY-OPTIMIZE)
                         (optimize-topology topology)
                         topology)
              storm-cluster-state (:storm-cluster-state nimbus)]
          (log-message "Received topology submission for " storm-name " with conf " storm-conf)
          (setup-storm-code conf storm-id uploadedJarLocation storm-conf topology)
          ;; protects against multiple storms being submitted at once and cleanup thread killing storm in b/w
          ;; assignment and starting the storm
          (locking (:submit-lock nimbus)
            (.setup-heartbeats! storm-cluster-state storm-id)
            (setup-storm-static conf storm-id storm-cluster-state)
            (mk-assignments conf storm-id storm-cluster-state
                            (fn [& ignored] (.add (:event-manager nimbus) reassign-fn))
                            (:task-heartbeats-cache nimbus))
            (start-storm storm-name storm-cluster-state storm-id))
          ))
      
      (^void killTopology [this ^String name]
        (.killTopologyWithOpts this name (KillOptions.)))

      (^void killTopologyWithOpts [this ^String storm-name ^KillOptions options]
        (check-storm-active! nimbus storm-name true)
        (let [storm-cluster-state (:storm-cluster-state nimbus)
              storm-id (get-storm-id storm-cluster-state storm-name)
              wait-amt (if (.is_set_wait_secs options)
                         (.get_wait_secs options)
                         ((read-storm-conf conf storm-id) TOPOLOGY-MESSAGE-TIMEOUT-SECS)
                         )]
          (set-topology-status! nimbus
                                storm-name
                                {:type :killed
                                 :kill-time-secs (+ (current-time-secs) wait-amt)})
          (.add (:cleanup-manager nimbus) cleanup-fn)
          ))

      (activate [this storm-name]
        (set-topology-status! nimbus
                              storm-name
                              {:type :active})
        )

      (deactivate [this storm-name]
        (set-topology-status! nimbus
                              storm-name
                              {:type :inactive})
        )

      (beginFileUpload [this]
        (let [fileloc (str (inbox nimbus) "/stormjar-" (uuid) ".jar")]
          (.put (:uploaders nimbus)
                fileloc
                (Channels/newChannel (FileOutputStream. fileloc)))
          (log-message "Uploading file from client to " fileloc)
          fileloc
          ))

      (^void uploadChunk [this ^String location ^ByteBuffer chunk]
        (let [uploaders (:uploaders nimbus)
              ^WritableByteChannel channel (.get uploaders location)]
          (when-not channel
            (throw (RuntimeException.
                    "File for that location does not exist (or timed out)")))
          (.write channel chunk)
          (.put uploaders location channel)
          ))

      (^void finishFileUpload [this ^String location]
        (let [uploaders (:uploaders nimbus)
              ^WritableByteChannel channel (.get uploaders location)]
          (when-not channel
            (throw (RuntimeException.
                    "File for that location does not exist (or timed out)")))
          (.close channel)
          (log-message "Finished uploading file from client: " location)
          (.remove uploaders location)
          ))

      (^String beginFileDownload [this ^String file]
        (let [is (BufferFileInputStream. file)
              id (uuid)]
          (.put (:downloaders nimbus) id is)
          id
          ))

      (^ByteBuffer downloadChunk [this ^String id]
        (let [downloaders (:downloaders nimbus)
              ^BufferFileInputStream is (.get downloaders id)]
          (when-not is
            (throw (RuntimeException.
                    "Could not find input stream for that id")))
          (let [ret (.read is)]
            (.put downloaders id is)
            (when (empty? ret)
              (.remove downloaders id))
            (ByteBuffer/wrap ret)
            )))

      (^String getTopologyConf [this ^String id]
        (to-json (read-storm-conf conf id)))

      (^StormTopology getTopology [this ^String id]
        (system-topology (read-storm-conf conf id) (read-storm-topology conf id)))


      ;; TODO: finish refactoring to new structure      

      
      (^ClusterSummary getClusterInfo [this]
        (let [storm-cluster-state (:storm-cluster-state nimbus)
              assigned (assigned-slots storm-cluster-state)
              supervisor-infos (all-supervisor-info storm-cluster-state)
              supervisor-summaries (dofor [[id info] supervisor-infos]
                                          (let [ports (set (:worker-ports info))
