         :compress-batch-ch compress-batch-ch
         :write-batch-ch write-batch-ch
         :status-check-ch status-check-ch
+        :ack-ch ack-ch
         :commit-tx-ch commit-tx-ch
-        :ack-ch ack-ch
         :close-resources-ch close-resources-ch
         :complete-task-ch complete-task-ch
         :reset-payload-node-ch reset-payload-node-ch

        java.lang.Exception
        (fn [e & _] (.printStackTrace e)))

      (dire/with-handler! #'commit-tx-loop
        java.lang.Exception
        (fn [e & _] (.printStackTrace e)))

      (dire/with-handler! #'ack-loop
        java.lang.Exception
        (fn [e & args] (.printStackTrace e)))

      (dire/with-handler! #'close-resources-loop
        java.lang.Exception

      (>!! status-check-ch (munge-write-batch event))
      (recur))))

(defn status-check-loop [status-ch ack-ch reset-payload-node-ch]
  (loop []
    (when-let [event (<!! status-ch)]
      (let [event (munge-status-check event)]
        (if (:commit? event)
          (>!! ack-ch event)
          (>!! reset-payload-node-ch event))
        (recur)))))

(defn ack-loop [ack-ch commit-ch]
  (loop []
    (when-let [event (<!! ack-ch)]
      (>!! commit-ch (munge-ack event))
      (recur))))

(defn commit-tx-loop [commit-ch close-resources-ch]
  (loop []
    (when-let [event (<!! commit-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))))

