             [cheshire.core :refer [parse-string]]
             [zookeeper :as zk]
             [onyx.static.default-vals :refer [arg-or-default]]
+            [onyx.types :as t]
             [onyx.peer.pipeline-extensions :as p-ext]
             [onyx.peer.function :as function]
             [onyx.peer.operation :refer [kw->fn]]

    [_ _ message-id]
    (when-let [msg (get @pending-messages message-id)]
      (swap! pending-messages dissoc message-id)
      (>!! read-ch msg)))

  (pending?
    [_ _ message-id]

                     (map (fn [_]
                            (let [result (first (alts!! [read-ch timeout-ch] :priority true))]
                              (if (= (:message result) :done)
                                (t/input (java.util.UUID/randomUUID) :done)
                                result))))
                     (filter :message))]
      (doseq [m batch]
        (swap! pending-messages assoc (:id m) m))
      (when (and (= 1 (count @pending-messages))
                 (= (count batch) 1)
                 (= (:message (first batch)) :done))
