simplify inbox cleaning code

Nathan Marz · Nathan Marz · commit 53ce477c78f9 · 2012-01-12T16:47:48.000-08:00
diff --git a/src/clj/backtype/storm/daemon/nimbus.clj b/src/clj/backtype/storm/daemon/nimbus.clj
@@ -617,6 +617,21 @@
           (swap! (:task-heartbeats-cache nimbus) dissoc id))
         ))))
 
+(defn- file-older-than? [now seconds file]
+  (<= (+ (.lastModified file) (to-millis seconds)) (to-millis now)))
+
+(defn clean-inbox [dir-location seconds]
+  "Deletes jar files in dir older than seconds."
+  (let [now (current-time-secs)
+        pred #(and (.isFile %) (file-older-than? now seconds %))
+        files (filter pred (file-seq (File. dir-location)))]
+    (doseq [f files]
+      (if (.delete f)
+        (log-message "Cleaning inbox ... deleted: " (.getName f))
+        ;; This should never happen
+        (log-error "Cleaning inbox ... error deleting: " (.getName f))
+        ))))
+
 (defn cleanup-corrupt-topologies! [nimbus]
   (let [storm-cluster-state (:storm-cluster-state nimbus)
         code-ids (set (code-ids (:conf nimbus)))
@@ -646,7 +661,7 @@
                         0
                         (conf NIMBUS-CLEANUP-FREQ-SECS)
                         (fn []
-                          (clean-inbox conf (inbox nimbus) (conf NIMBUS-INBOX-JAR-EXPIRATION-SECS))
+                          (clean-inbox (inbox nimbus) (conf NIMBUS-INBOX-JAR-EXPIRATION-SECS))
                           ))
     (reify Nimbus$Iface
       (^void submitTopology
@@ -865,34 +880,12 @@
              (FileUtils/copyFile src-file (File. (master-stormjar-path stormroot)))
              ))
 
-(defn- file-older-than? [now seconds file]
-  (<= (+ (.lastModified file) (* seconds 1000)) now))
-
-(defn- is-jar? [f]
-  (-> f .getName (.endsWith ".jar")))
-
-(defmethod clean-inbox :distributed [conf dir-location seconds]
-  "Deletes jar files in dir older than seconds."
-  (let [now (Time/currentTimeMillis)
-        pred #(and (is-jar? %) (file-older-than? now seconds %))
-        files (filter pred (file-seq (File. dir-location)))]
-    (doseq [f files]
-      (if (.delete f)
-        (log-message "Cleaning inbox ... deleted: " (.getName f))
-        ;; This should never happen
-        (log-error "Cleaning inbox ... error deleting: " (.getName f))
-        ))))
-
 ;; local implementation
 
 (defmethod setup-jar :local [conf & args]
   nil
   )
 
-(defmethod clean-inbox :local [conf & args]
-  nil
-  )
-
 
 (defn -main []
   (launch-server! (read-storm-config)))
diff --git a/test/clj/backtype/storm/nimbus_test.clj b/test/clj/backtype/storm/nimbus_test.clj
@@ -451,15 +451,15 @@
        (doseq [fs [["a.jar" 20] ["b.jar" 20] ["c.jar" 0]]]
          (apply mk-file fs))
        (assert-files-in-dir ["a.jar" "b.jar" "c.jar"])
-       (nimbus/clean-inbox {STORM-CLUSTER-MODE "distributed"} dir-location 10)
+       (nimbus/clean-inbox dir-location 10)
        (assert-files-in-dir ["c.jar"])
        ;; Cleanit again, c.jar should stay
        (advance-time-secs! 5)
-       (nimbus/clean-inbox {STORM-CLUSTER-MODE "distributed"} dir-location 10)
+       (nimbus/clean-inbox dir-location 10)
        (assert-files-in-dir ["c.jar"])
        ;; Advance time, clean again, c.jar should be deleted.
        (advance-time-secs! 5)
-       (nimbus/clean-inbox {STORM-CLUSTER-MODE "distributed"} dir-location 10)
+       (nimbus/clean-inbox dir-location 10)
        (assert-files-in-dir [])
        )))
 
