trigger/emit example problem

basic.clj

(ns onyx-trigger-problem.jobs.basic
  (:require [onyx.job :refer [add-task register-job]]
            [onyx.tasks.core-async :as core-async-task]
            [onyx-trigger-problem.tasks.math :as math]))

(defn batch-done?
  [trigger {:keys [event-type segment] :as state-event}]
  (= event-type :job-completed))

(defn emit-inc
  [event window trigger {:keys [lower-bound upper-bound event-type] :as state-event} extent-state]
  (when extent-state
    (let [new-extent-state [{:n (reduce #(+ %1 (:n %2)) 0 extent-state)}]]
      (println extent-state)
      (println new-extent-state)
      new-extent-state)))

(defn basic-job
  [batch-settings]
  (let [base-job {:workflow        [[:in :inc]
                                    [:inc :out]]
                  :catalog         []
                  :lifecycles      []
                  :windows         [{:window/id          :window-inc
                                     :window/task        :inc
                                     :window/type        :global
                                     :window/aggregation :onyx.windowing.aggregation/conj}]
                  :triggers        [{:trigger/window-id         :window-inc
                                     :trigger/id                :punctuation-trigger
                                     :trigger/refinement        :onyx.refinements/discarding
                                     :trigger/fire-all-extents? true
                                     :trigger/on                :onyx.triggers/punctuation
                                     :trigger/pred              ::batch-done?
                                     :trigger/emit              ::emit-inc}
                                    {:trigger/window-id         :window-inc
                                     :trigger/id                :segment-trigger
                                     :trigger/refinement        :onyx.refinements/discarding
                                     :trigger/fire-all-extents? true
                                     :trigger/on                :onyx.triggers/segment
                                     :trigger/threshold         [5 :elements]
                                     :trigger/emit              ::emit-inc}]
                  :flow-conditions []
                  :task-scheduler  :onyx.task-scheduler/balanced}]
    (-> base-job
        (add-task (core-async-task/input :in batch-settings))
        (add-task (math/inc-key :inc [:n] batch-settings))
        (add-task (core-async-task/output :out batch-settings)))))

(defmethod register-job "basic-job"
  [job-name config]
  (let [batch-settings {:onyx/batch-size 1 :onyx/batch-timeout 1000}]
    (basic-job batch-settings)))