 ; This is the channel for queuing all user requests
 (def all-collector-chan (chan (sliding-buffer collector-chan-size)))
 
+(defn- build-user-connector-msg
+  [user connector]
+  (str "user " (get-name user) " on cloud " (.getConnectorInstanceName connector)))
+
+(defn- build-msg
+  [prefix user connector suffix]
+  (str prefix "for " (build-user-connector-msg user connector) suffix))
+
+(defn log-timeout
+  [user connector]
+  (log/log-error (build-msg "Timed out in waiting for collecting vms " user connector "")))
+
+(defn log-failure
+  [user connector]
+  (log/log-error (build-msg "Failed collecting vms " user connector "")))
+
+(defn log-collected
+  [user connector v]
+  (log/log-info (build-msg "Number of VMs collected " user connector (str ": " v))))
+
 (defn collect!
   [user connector]
   (let [ch (chan 1)]
     (go
-      (let [[v c] (alts! [ch (timeout timeout-collect)])]
-        (if (nil? v)
-          (log/log-error
-            "Timeout collecting vms for user "
-            (get-name user)
-            " on cloud "
-            (.getConnectorInstanceName connector))
-          (log/log-info "Number of VMs collected for user " (get-name user)
-            " on cloud "
-            (.getConnectorInstanceName connector) ": " v))))
-    (go (>! ch (Collector/collect user connector)))))
+      (let [[v _] (alts! [ch (timeout timeout-collect-reader-release)])]
+        (swap! busy free [(get-name user) connector])
+        (cond
+          (nil? v) (log-timeout user connector)
+          (< v 0) (log-failure user connector)
+          :else (log-collected user connector v))))
+    (go (>! ch (Collector/collect user connector (msecs-in-seconds timeout-collect))))))
 
 (defn update-metric!
   [user]

      (while true
        (let [[[user connector] ch] (alts! [chan (timeout timeout-processing-loop)])]
          (when (not-nil? user)
            (try
              (log/log-debug (str "executing collect request for " (get-name user) " and " (.getConnectorInstanceName connector)))
              (collect! user connector)
              (if (updator/metering-enabled?)
                (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]
      (if-not (@busy [(get-name u) c])
        (go (>! chan [u c]))
        (log/log-info "Avoiding working on " (build-user-connector-msg u c) " being in a process." )))))

; 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
  []
