     (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))
-    (close! (:reset-payload-node-ch component))
 
     component))
 

        :status-check-loop (thread (status-check-loop status-check-ch ack-ch reset-payload-node-ch))
        :ack-loop (thread (ack-loop ack-ch commit-tx-ch))
        :commit-tx-loop (thread (commit-tx-loop commit-tx-ch close-resources-ch))
        :close-resources-loop (thread (close-resources-loop close-resources-ch complete-task-ch))
        :complete-task-loop (thread (complete-task-loop complete-task-ch reset-payload-node-ch))
        :reset-payload-node (thread (reset-payload-node reset-payload-node-ch)))))

  (stop [component]
    (prn "Stopping Task Pipeline")

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

(defn close-resources-loop [close-ch reset-payload-ch]
  (loop []
    (when-let [event (<!! close-ch)]
      (>!! reset-payload-ch (munge-close-resources event))
      (recur))))

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

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

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