                                                                 (count ports)
                                                                 (count (assigned id)))
                                             ))
-              nimbus-uptime (uptime)
+              nimbus-uptime ((:uptime nimbus))
               bases (topology-bases storm-cluster-state)
               topology-summaries (dofor [[id base] bases]
                                         (let [assignment (.assignment-info storm-cluster-state id nil)]

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

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

(defserverfn service-handler [conf]
  (let [submitted-count (atom 0)
        active (atom true)
        conf (merge (read-storm-config) conf) ;; useful when testing
        storm-cluster-state (cluster/mk-storm-cluster-state conf)
        [event-manager cleanup-manager :as managers] [(event/event-manager false) (event/event-manager false)]
        inbox (master-inbox conf)
        storm-submit-lock (Object.)
        task-heartbeats-cache (atom {}) ; map from storm id -> task id -> {:nimbus-time :task-reported-time}
        downloaders (file-cache-map conf)
        uploaders (file-cache-map conf)
        uptime (uptime-computer)
        
        cleanup-fn (fn []
                     (let [id->kill-time (locking storm-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 storm-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 dissoc id)
                            )
                          (log-message "Cleaned up topology task heartbeats: " (pr-str to-cleanup-ids))
                          )))
        reassign-fn (fn this []
                      (when (conf NIMBUS-REASSIGN)
                        (locking storm-submit-lock
                          (let [callback (fn [& ignored] (.add event-manager 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))))
                              ))))
        threads [(async-loop
                   (fn []
                     (.add event-manager reassign-fn)
                     (.add cleanup-manager cleanup-fn)
                     (when @active (conf NIMBUS-MONITOR-FREQ-SECS))
                     ))
                   ]]

    (reify Nimbus$Iface
      (^void submitTopology
             [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology]
             (when (storm-active? storm-cluster-state storm-name)
               (throw (AlreadyAliveException. storm-name)))
             (validate-topology! topology)
             (swap! submitted-count inc)
             (let [storm-id (str storm-name "-" @submitted-count "-" (current-time-secs))
                   storm-conf (from-json serializedConf)
                   storm-conf (assoc storm-conf STORM-ID storm-id)
                   storm-conf (normalize-conf conf storm-conf)
                   total-storm-conf (merge conf storm-conf)
                   topology (if (total-storm-conf TOPOLOGY-OPTIMIZE) (optimize-topology topology) topology)]
               (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 storm-submit-lock
                 (.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 reassign-fn)) task-heartbeats-cache)
                 (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]
        (letlocals
         (bind storm-id
               (get-storm-id storm-cluster-state storm-name))         
         (when-not storm-id
           (throw (NotAliveException. storm-name)))
         (bind 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! storm-cluster-state
                               storm-submit-lock
                               storm-name
                               {:type :killed
                                :kill-time-secs (+ (current-time-secs) wait-amt)})
         (.add cleanup-manager cleanup-fn)
         ))

      (activate [this storm-name]
        (set-topology-status! storm-cluster-state
                              storm-submit-lock
                              storm-name
                              {:type :active})
        )

      (deactivate [this storm-name]
        (set-topology-status! storm-cluster-state
                              storm-submit-lock
                              storm-name
                              {:type :inactive})
        )

      (beginFileUpload [this]
        (let [fileloc (str inbox "/stormjar-" (uuid) ".jar")]
          (.put uploaders fileloc (Channels/newChannel (FileOutputStream. fileloc)))
          (log-message "Uploading file from client to " fileloc)
          fileloc
          ))
      
      (^void uploadChunk [this ^String location ^ByteBuffer chunk]
             (let [^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 [^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 id is)
                 id
                 ))

      (^ByteBuffer downloadChunk [this ^String id]
              (let [^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)))
      
      (^ClusterSummary getClusterInfo [this]
        (let [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))

;;    -- swapped + active other = wait + inactivate other
;;    -- inactive other + swapped = wait message timeout + kill(0)
;;    -- swapped + killed other = activate
;; -- rebalance, :wait-time
;;      -- after waiting, should compute new assignment from scratch

;; swapping design
;; -- need 2 ports per worker (swap port and regular port)
