 (ns onyx.plugin.datomic
-  (:require [clojure.core.async :refer [chan >! >!! <!! close! go timeout alts!!]]
+  (:require [clojure.core.async :refer [chan >! >!! <!! close! thread timeout alts!! go-loop sliding-buffer]]
             [datomic.api :as d]
             [onyx.peer.pipeline-extensions :as p-ext]
             [onyx.peer.function :as function]

        max-pending (or (:onyx/max-pending catalog-entry) (:onyx/max-pending defaults))
        batch-size (:onyx/batch-size catalog-entry)
        batch-timeout (or (:onyx/batch-timeout catalog-entry) (:onyx/batch-timeout defaults))
        read-ch (chan (or (:datomic/read-buffer catalog-entry) 1000))] 
    (->DatomicInput (:onyx.core/log pipeline-data)
                    (:onyx.core/task-id pipeline-data)
                    max-pending batch-size batch-timeout 

(defrecord DatomicInput [log task-id max-pending batch-size batch-timeout 
                         pending-messages drained?
                         top-chunk-index top-acked-chunk-index pending-chunk-indices 
                         read-ch commit-ch]
  p-ext/Pipeline
  (write-batch 
    [this event]
