         :ack-ch ack-ch
         :commit-tx-ch commit-tx-ch
         :close-resources-ch close-resources-ch
+        :reset-payload-node-ch reset-payload-node-ch
         :complete-task-ch complete-task-ch
-        :reset-payload-node-ch reset-payload-node-ch
         
         :open-session-loop (thread (open-session-loop read-batch-ch open-session-kill-ch pipeline-data))
         :read-batch-loop (thread (read-batch-loop read-batch-ch decompress-batch-ch))

      (>!! close-resources-ch (munge-commit-tx event))
      (recur))))

(defn close-resources-loop [close-ch complete-task-ch]
  (loop []
    (when-let [event (<!! close-ch)]
      (taoensso.timbre/info "Closing out resources for " (:session event))
      (>!! complete-task-ch (munge-close-resources event))
      (recur))))

(defn complete-task-loop [complete-ch reset-payload-node-ch]
  (loop []
    (when-let [event (<!! complete-ch)]
      (let [event (munge-complete-task event)]
        (when (:tail-batch? event)
          (>!! reset-payload-node-ch event)))
      (recur))))

(defn reset-payload-node [reset-ch]
  (when-let [event (<!! reset-ch)]
    (munge-new-payload event)
    (>!! (:complete-ch event) true)))

(defrecord TaskPipeline [payload sync queue payload-ch complete-ch]
  component/Lifecycle

          ack-ch (chan 1)
          commit-tx-ch (chan 1)
          close-resources-ch (chan 1)
          reset-payload-node-ch (chan 1)
          complete-task-ch (chan 1)

          pipeline-data {:ingress-queues (:task/ingress-queues (:task payload))
                         :egress-queues (:task/egress-queues (:task payload))
