     (loop [items items num-items 1]
       (when (and items (not (closed? chan)))
 
-        (if-let [item (first items)]
+        (when-let [item (first items)]
           (push chan item)
 
           ;; Intermediate logging


(defn consume-all
  "Consumes all the messages on the channel with the function f. Continues to wait until channel
  returns nil or stop-fn returns true for an item."
  [chan f]
  (let [start (System/currentTimeMillis)]
    (loop [item (pull chan) num-items 1]
      (when item

        ;; process the item
        (f item)


  (pull c)
  (count (:queue c))

  (process-tasks :tasks-fn #(range 4000)
                 :processor (fn [t]
                              (info "processing " t)
                              (Thread/sleep 10))
                 :num-threads 4
                 :task-name "sleep-test")

  (future
    (push-seq c (range 100))
    (println "Finished pushing 100 items")
