gistfile1.txt

;;
;; Workers
;;

(defn worker-start
  [cfg topic func]
  (let [p (promise)]
    {:p p
     :f (future
          (with-consumer c cfg
            (consumer-subscribe c topic)
            (while (not (realized? p))
              (when-let [msgs (not-empty (consumer-poll c))]
                (pmap func msgs)))))}))

(defn worker-stop
  [w]
  (deliver (:p w) true))

(defn worker-status
  [w]
  (and
   (not (future-cancelled? (:f w)))
   (not (realized? (:f w)))))

(defn worker-error
  [w]
  (when-not (worker-status w)
    (try
      (-> w :f deref)
      nil
      (catch Exception e
        e))))