   (reduce + (map :messages (filter-by-rkmap rkmap (filter-by-vhost vhost queue-state)))))
 
 
-(defn wait-for-zero-messages
-  [get-num-messages]
-  (loop  [zero-count 0]
-    (if (= 0 (get-num-messages))
-      (when (< zero-count 6)
-        (Thread/sleep 500)
-        (recur (inc zero-count)))
-      (do
-        (Thread/sleep 10000)
-        (recur 0)))))
-
-
 (defn start-synchronization
   [rmq-settings id filesystem]
   (let [conn (rmq/connect rmq-settings)

    (make-object-from-section [this system section-name]
      (let [rmq-settings (-> system :config :rmq-settings)
            rmq-prefix (-> system :config :rmq-prefix)
            filesystem-names (sys/get-filesystems-for-section system section-name)
            filesystems (map #(vfs/make-additional-fs-map system %) filesystem-names)]
        (http-connect! rmq-settings)
        (->ManageService rmq-settings rmq-prefix filesystems)))))

    (rmq/close ch)
    (rmq/close conn)))

(defn wait-idle
  [get-num-messages]
  (go
   (loop []
     (if-let [n (<! (get-num-messages))]
       (if (zero? n)
         :idle
         (recur))))))

(defn manage-filesystem
  [rmq-settings id {:keys [fsid sleep-between-sync]} ch]
  (let [qname (rk/routing-key-string {:id id :filesystem fsid :command "*"})
        get-num-messages* #(go
                            (if-let [qstate (<! ch)]
                              (num-messages-from-queue-state qstate
                                                             (:vhost rmq-settings)
                                                             {:id  id :filesystem fsid})))]
    (go
     (loop []
       (when-let [num-messages (<! (get-num-messages*))]
         (if (zero? num-messages)
           (do
             (logging/info "starting synchronization of" qname)
             (<! (thread (start-synchronization rmq-settings id fsid)))
             ;; consume the next value since that may have been
             ;; produced before the thread was started
             (<! ch))
           (logging/info "synchronization of" qname "already running"))
         (<! (wait-idle get-num-messages*))
         (logging/info "synchronization of" qname "finished. restarting in" sleep-between-sync "seconds")
         (<! (timeout (* sleep-between-sync 1000)))
         (recur))))))


(defrecord ManageService [rmq-settings rmq-prefix filesystems qwatch-mult]
  worker/Service
  (start [this]
    (doseq [fs filesystems]
      (let [ch (chan (async/sliding-buffer 1))]
        (async/tap qwatch-mult ch)
        (manage-filesystem rmq-settings rmq-prefix fs ch)))))


(def runner
  (reify
