              basis-t (d/as-of basis-t))]
     {:datomic/conn conn
      :datomic/db db
-     :onyx.core/params (if param? 
+     :onyx.core/params (if param?
                          (conj params db)
                          params)}))
 

  (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)}))


          batch (->> (range max-segments)
                     (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true)))))]
      (doseq [m batch]
        (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})
      (reset! drained? true))
    {:onyx.core/batch batch}))

