                                    :lifecycle/calls ::log-calls
                                    :lifecycle/doc "Lifecycle for printing the output of a task's batch"}]}))
 
-;;;;================- Core.Async  -===========================
+;;;;======================================================
+;;;;                 Core Async
 (def get-input-channel
   "Returns the same channel every time for an id."
   (memoize

              [cheshire.core :as json]))


;;;;================- Logging -===========================

(defn log-batch [event lifecycle]
  (let [task-name (:onyx/name (:onyx.core/task-map event))]
    (doseq [m (map :message (mapcat :leaves (:tree (:onyx.core/results event))))]

                                :core.async/id (or uuid (java.util.UUID/randomUUID))}
                               {:lifecycle/task output-task-name
                                :lifecycle/calls :onyx.plugin.core-async/writer-calls}])])})))

;;;;======================================================
;;;;                 Kafka
(defn deserialize-message [bytes]
  (try
    (json/parse-string (String. bytes "UTF-8"))
