   (if (= :input (:onyx/type task-map))
     (p/start (instantiate-plugin event) event)))
 
-(defn build-output-pipeline [{:keys [onyx.core/task-map] :as event}]
+(defn build-output-pipeline [event]
   (cond (source-task? event) 
         (p/start (mo/new-messenger-output event) event)
 

    (advance state)))

(defn trigger-checkpointed-state-event [state epoch]
  (let [messenger (get-messenger state)
        {:keys [onyx.core/triggers] :as event} (get-event state)]
    (if (not (empty? triggers)) 
      (ws/assign-windows state
                         (onyx.types/checkpointed-state-event 

        state)
      identity)))

(defn build-read-batch [{:keys [onyx.core/task-map onyx.core/since-barrier-count] :as event}] 
  (let [batch-timeout-ns (long (ms->ns (arg-or-default :onyx/batch-timeout task-map)))
        batch-size (long (:onyx/batch-size task-map))] 
    (if (source-task? event) 
