 (defn add-completion-id [event m]
   (assoc m :completion-id (:onyx.core/id event)))
 
-(defn tag-each-message [event]
-  (if (= (:onyx/type (:onyx.core/task-map event)) :input)
-    (let [event (update-in event 
-                           [:onyx.core/batch]
-                           (fn [batch]
-                             (map (comp (partial add-completion-id event)
-                                        (partial add-acker-id event)
-                                        add-ack-value)
-                                  batch)))]
-      (doseq [raw-segment (:onyx.core/batch event)]
-        (extensions/internal-ack-message
-         (:onyx.core/messenger event)
-         event
-         (operation/peer-link event (:acker-id raw-segment) :acker-peer-site)
-         (:id raw-segment)
-         (:completion-id raw-segment)
-         (:ack-val raw-segment)))
-      event)
-    event))
-
 (defn sentinel-found? [event]
   (seq (filter (partial = :done) (map :message (:onyx.core/decompressed event)))))
 

        (= to-add :none) #{}
        :else (clojure.set/union (into #{} all) (into #{} to-add))))

(defn choose-output-paths [flow-conditions event new downstream]
  (if (seq flow-conditions)
    (reduce
     (fn [{:keys [paths exclusions] :as all} entry]
       (if ((:flow/predicate entry) [event new])
         (if (:flow/short-circuit? entry)
           (reduced {:paths (join-output-paths paths (:flow/to entry) downstream)
                     :exclusions (clojure.set/union (into #{} exclusions) (into #{} (:flow/exclude-keys entry)))})
           {:paths (join-output-paths paths (:flow/to entry) downstream)
            :exclusions (clojure.set/union (into #{} exclusions) (into #{} (:flow/exclude-keys entry)))})
         all))
     {:paths #{} :exclusions #{}}
     flow-conditions)
    {:paths downstream}))

(defn output-paths
  [{:keys [onyx.core/serialized-task onyx.core/compiled-flow-conditions] :as event}]
  (merge
   event
   {:onyx.core/result-paths
    (map
     (fn [segment]
       (choose-output-paths compiled-flow-conditions event segment
                            (keys (:egress-ids serialized-task))))
     (:onyx.core/results event))}))

(defn inject-batch-resources [event]
  (let [cycle-params {:onyx.core/lifecycle-id (java.util.UUID/randomUUID)}]
    (merge event cycle-params (l-ext/inject-batch-resources* event))))

(defn read-batch [event]
  (let [rets (tag-each-message (merge event (p-ext/read-batch event)))]
    (when (= (:onyx/type (:onyx.core/task-map event)) :input)
      (doseq [m (:onyx.core/batch rets)]
        (go (try (<! (timeout (or (:onyx/pending-timeout (:onyx.core/task-map event)) 60000)))
                 (when (p-ext/pending? rets (:id m))
                   (p-ext/retry-message event (:id m)))
                 (catch Exception e
                   (taoensso.timbre/warn e))))))
    rets))

(defn decompress-batch [event]
  ;; TODO: Messages are already decompressed, need to rework the interface.
  (let [rets (merge event {:onyx.core/decompressed (:onyx.core/batch event)})]
    (if (sentinel-found? rets)
      (do (if (p-ext/drained? rets)
            (complete-job rets)
            (p-ext/retry-message rets (sentinel-id rets)))
          (strip-sentinel rets))
      rets)))

(defn apply-fn* [{:keys [onyx.core/batch onyx.core/decompressed] :as event} batch-f single-f]
  (if (seq decompressed)
    (let [batch-result (batch-f event)
          rets
          (merge event
                 (reduce
                  (fn [rets [raw thawed]]
                    (let [new-segments (single-f event thawed batch-result)
                          new-ack-vals (map (fn [x] (acker/gen-ack-value)) new-segments)
                          tagged (apply acker/prefuse-vals new-ack-vals)
                          results (map (fn [segment ack-val]
                                         {:id (:id raw)
                                          :acker-id (:acker-id raw)
                                          :completion-id (:completion-id raw)
                                          :message segment
                                          :ack-val ack-val})
                                       new-segments new-ack-vals)]
                      (-> rets
                          (update-in [:onyx.core/results] concat results)
                          (assoc-in [:onyx.core/children raw] tagged))))
                  {:onyx.core/results [] :onyx.core/children {}}
                  (map vector batch decompressed)))]
      (ack-messages rets)
      (merge rets (output-paths rets)))
    (merge event {:onyx.core/results []})))

(defn apply-fn [event]
  (if (:onyx/batch? (:onyx.core/task-map event))
    (apply-fn*
     event
     (fn [{:keys [onyx.core/decompressed] :as event}]
       (let [new-segments (p-ext/apply-fn event (map :message decompressed))]
         (if (sequential? new-segments) new-segments (vector new-segments))))
     (fn [event thawed batch-result]
       batch-result))
    (apply-fn*
     event
     (fn [_] nil)
     (fn [event thawed batch-result]
       (let [new-segments (p-ext/apply-fn event (:message thawed))]
         (if (sequential? new-segments) new-segments (vector new-segments)))))))

(defn compress-batch [event]
  (merge event (p-ext/compress-batch event)))

    (acker/prefuse-vals parent-ack child-ack)))

(defn ack-messages [{:keys [onyx.core/acking-daemon onyx.core/children] :as event}]
  (merge
   event
   (if children
     (do
       (doseq [raw-segment (keys children)]
         (when (:ack-val raw-segment)
           (let [link (operation/peer-link event (:acker-id raw-segment) :acker-peer-site)]
             (extensions/internal-ack-message
              (:onyx.core/messenger event)
              event
              link
              (:id raw-segment)
              (:completion-id raw-segment)
              (fuse-ack-vals (:onyx.core/task-map event) (:ack-val raw-segment) (get children raw-segment))))))
       {})
     {})))

(defn segments->ack-values [event segments]
  (if (= :output (:onyx/type (:onyx.core/task-map event)))
