                          :completion-node (:completion (:nodes payload))
                          :catalog (read-string (extensions/read-place sync (:catalog (:nodes payload))))
                          :workflow (read-string (extensions/read-place sync (:workflow (:nodes payload))))
+                         :payload-ch payload-ch
                          :queue queue
                          :sync sync
                          :batch-size 2

          (>!! reset-payload-node-ch event)))
      (recur))))

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


        :commit-tx-loop (future (commit-tx-loop commit-tx-ch close-resources-ch))
        :close-resources-loop (future (close-resources-loop close-resources-ch complete-task-ch))
        :complete-task-loop (future (complete-task-loop complete-task-ch reset-payload-node-ch))
        :reset-payload-node (future (reset-payload-node reset-payload-node-ch)))))

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