       (while true
         (let [[[user connector] ch] (alts! [chan (timeout timeout-processing-loop)])]
           (when (not-nil? user)
+            (swap! busy mark-busy [(get-name user) connector])
             (try
-              (log/log-debug (str "executing collect request for " (get-name user) " and " (.getConnectorInstanceName connector)))
+              (log/log-info (str "executing collect request for " (get-name user) " and " (.getConnectorInstanceName connector)))
               (collect! user connector)
               (if (updator/metering-enabled?)
-                (update-metric! user))
+                 (update-metric! user))
               (catch Exception e (log/log-warn "caught exception executing collect request: " (.getMessage e))))))))))
 
 (defonce ^:dynamic *online-collect-processor* (collect-readers online-collector-chan))

  (let [connectors (connectors)]
    (check-channel-size users connectors)
    (doseq [u users
            c connectors]
      (go
        (>! chan [u c])))))

; Start collector writers
(defn collect-writers
  []
  (go
	  (while true
	    (<! (timeout timeout-online-loop))
	    (let [users (online-users)]
	      (insert-collection-requests users online-collector-chan))))
  (go
	  (while true
	    (<! (timeout timeout-all-users-loop))
	    (let [users (users)]
	      (insert-collection-requests users all-collector-chan)))))

(defn -start
  []

    :name slipstream.async.Collector
    :methods [#^{:static true
                 :doc "Takes: run user"}
[start [] void]]))

(defn seconds-in-msecs
  [seconds]
  (* 1000 seconds))

(defn msecs-in-seconds
  [msecs]
  (/ msecs 1000))

(def busy (atom #{}))

(defn- mark-busy
 [current-busy v]
 (conj current-busy v))

(defn- free
 [current-busy v]
 (disj current-busy v))

(def collector-chan-size 512)
(def number-of-readers 32)
(def timeout-all-users-loop (seconds-in-msecs 240))
(def timeout-online-loop (seconds-in-msecs 10))
(def timeout-collect (* 2 timeout-collect-loop))
(def timeout-collect-reader-release (+ timeout-collect (seconds-in-msecs 2)))
(def timeout-processing-loop (seconds-in-msecs 60))

(defn get-value
