 (ns onyx.peer.task-pipeline
   (:require [clojure.core.async :refer [alts!! <!! >!! chan close!]]
             [com.stuartsierra.component :as component]
+            [dire.core :as dire]
             [onyx.queue.hornetq :refer [hornetq]]
             [onyx.extensions :as extensions]))
 

                         :sync sync
                         :batch-size 1000
                         :timeout 50}]

      (assoc component
        :read-batch-ch read-batch-ch

  (extensions/create-tx-session queue))

(defn read-batch [queue consumers batch-size timeout]
  (let [consumer-chs (map (fn [_] (chan 1)) (range (count consumers)))]
    (doseq [[consumer ch] (map vector consumers consumer-chs)]
      (>!! ch (extensions/consume-message queue consumer timeout)))
    (let [rets (->> (range batch-size)
                    (map (fn [_] (first (apply alts!! consumer-chs))))
