     (when (and (= 1 (count @pending-messages))
                (= (count batch) 1)
                (= (:message (first batch)) :done))
-      (extensions/force-write-chunk log :chunk {:status :complete} task-id)
+      (>!! commit-ch {:status :complete}) 
       (reset! drained? true))
     {:onyx.core/batch batch}))
 

(defn safe-datoms-per-segment [task-map]
  (or (:datomic/datoms-per-segment task-map)
      (throw (ex-info ":datomic/datoms-per-segment missing from write-datoms task-map." task-map))))

(defn inject-read-datoms-resources
  [{:keys [onyx.core/task-map onyx.core/log onyx.core/task-id onyx.core/pipeline] :as event} lifecycle]

(ns onyx.plugin.datomic
  (: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]
