           batch (->> (range max-segments)
                      (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true)))))]
       (doseq [m batch]
-        (when-let [chunk-index (:chunk-index m)] 
+        (when-let [chunk-index (:chunk-index m)]
           (swap! top-chunk-index max chunk-index)
           (swap! pending-chunk-indices conj chunk-index))
         (swap! pending-messages assoc (:id m) m))
     (when (and (= 1 (count @pending-messages))
                (= (count batch) 1)
                (= (:message (first batch)) :done))
-      (>!! commit-ch {:status :complete}) 
+      (>!! commit-ch {:status :complete})
       (reset! drained? true))
     {:onyx.core/batch batch}))
 


(defrecord DatomicWriteDatoms [conn partition]
  p-ext/Pipeline
  (read-batch 
    [_ event]
    (function/read-batch event))

  (write-batch 
    [_ event]
    (let [messages (mapcat :leaves (:tree (:onyx.core/results event)))]
      @(d/transact conn
                   (map (fn [msg] (if (and partition (not (sequential? msg))) 
                                    (assoc msg :db/id (d/tempid partition))
                                    msg))
                        (map :message messages)))
      {:onyx.core/written? true}))

  (seal-resource 
    [_ _]
    {}))

(defn write-datoms [pipeline-data]
  (let [task-map (:onyx.core/task-map pipeline-data)
        conn (safe-connect task-map) 
        partition (:datomic/partition task-map)] 
    (->DatomicWriteDatoms conn partition)))

(defrecord DatomicWriteBulkDatoms [conn]
  p-ext/Pipeline
  (read-batch 
    [_ event]
    (function/read-batch event))

  (write-batch 
    [_ event]
    ;; Transact each tx individually to avoid tempid conflicts.
    (doseq [tx (mapcat :leaves (:tree (:onyx.core/results event)))]
      @(d/transact conn (:tx (:message tx))))
    {:onyx.core/written? true})

  (seal-resource 
    [_ _]
    {}))

(defn write-bulk-datoms [pipeline-data]
  (let [task-map (:onyx.core/task-map pipeline-data)
        conn (safe-connect task-map)] 
    (->DatomicWriteBulkDatoms conn)))

(def read-datoms-calls

          range-end (:datomic/index-range-end task-map)]
      (d/index-range db attribute range-start range-end))))

(defn close-read-datoms-resources
  [{:keys [datomic/producer-ch datomic/commit-ch datomic/read-ch] :as event} lifecycle]
  (close! read-ch)
  (close! commit-ch)
