           range-end (:datomic/index-range-end task-map)]
       (d/index-range db attribute range-start range-end))))
 
-(defn close-read-datoms-resources 
+(defn close-read-datoms-resources
   [{:keys [datomic/producer-ch datomic/commit-ch datomic/read-ch] :as event} lifecycle]
   (close! read-ch)
   (close! commit-ch)

      max-pending
      (recur (inc max-pending)))))

(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]
    (function/write-batch event))

  (read-batch 
    [_ event]
    (let [pending (count (keys @pending-messages))
          max-segments (min (- max-pending pending) batch-size)

  (when-not uri
    (throw (ex-info "Missing :datomic/uri in inject-conn-calls lifecycle."
                    lifecycle)))
  (let [conn (d/connect uri)]
    {:datomic/conn conn
     :onyx.core/params (if param?
                         (conj params conn)
                         params)}))

