         java.lang.Exception
         (fn [e & _] (.printStackTrace e)))
 
+      (dire/with-handler! #'reset-payload-node
+        java.lang.Exception
+        (fn [e & _] (.printStackTrace e)))
+
       (dire/with-handler! #'complete-task-loop
-        java.lang.Exception
-        (fn [e & _] (.printStackTrace e)))
-
-      (dire/with-handler! #'reset-payload-node
         java.lang.Exception
         (fn [e & _] (.printStackTrace e)))
 

      (>!! 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

    (close! (:ack-ch component))
    (close! (:commit-tx-ch component))
    (close! (:close-resources-ch component))
    (close! (:reset-payload-node-ch component))
    (close! (:complete-task-ch component))

    component))

