 ;; output plugins
 
 (defn inject-write-tx-resources
-  [{:keys [onyx.core/pipeline]} lifecycle]
+  [{:keys [onyx.core/pipeline onyx.core/task-map]} lifecycle]
   {:datomic/conn (:conn pipeline)})
 
 (defn inject-write-bulk-tx-resources

  (when-not (or (= 1 (:onyx/max-peers task-map))
                (= 1 (:onyx/n-peers task-map)))
    (throw (ex-info "Read datoms tasks must set :onyx/max-peers 1" task-map)))
  (let [_ (extensions/write-chunk log :chunk {:chunk-index -1 :status :incomplete} task-id)
        content (extensions/read-chunk log :chunk task-id)]
    (if (= :complete (:status content))

(defn datoms-sequence [db task-map]
  (case (:onyx/plugin task-map)
    ::read-datoms
    (let [_ (s/validate DatomicReadDatomsTaskMap task-map)
          datoms-components (or (:datomic/datoms-components task-map) [])
          datoms-index (:datomic/datoms-index task-map)]
      (apply d/datoms db datoms-index datoms-components))
    ::read-index-range
    (let [_ (s/validate DatomicReadIndexRangeTaskMap task-map)
          attribute (:datomic/index-attribute task-map)
          range-start (:datomic/index-range-start task-map)
          range-end (:datomic/index-range-end task-map)]
      (d/index-range db attribute range-start range-end))))
