                            ^int iteration-idx 
                            ^int batch-idx
                            ^int nstates 
-                           lifecycle-names
+                           #^"[Lclojure.lang.Keyword;" lifecycle-names
                            #^"[Lclojure.lang.IFn;" lifecycle-fns 
                            ^:unsynchronized-mutable ^int idx 
                            ^:unsynchronized-mutable ^java.lang.Boolean advanced 
                            ^:unsynchronized-mutable sealed
                            ^:unsynchronized-mutable replica 
+                           monitoring 
                            ^:unsynchronized-mutable messenger 
                            messenger-group
                            ^:unsynchronized-mutable coordinator

            [onyx.static.logging :as logger]
            [onyx.state.state-extensions :as state-extensions]
            [onyx.static.util :refer [ms->ns]]
            [onyx.types :refer [->Results ->MonitorEvent]]
            [schema.core :as s]
            [taoensso.timbre :refer [debug info error warn trace fatal]])
  (:import [org.agrona.concurrent IdleStrategy SleepingIdleStrategy]

(defn new-task-information [peer task]
  (map->TaskInformation (select-keys (merge peer task) [:log :job-id :task-id :id])))

(defn build-lifecycles [event]
  [{:lifecycle :lifecycle/poll-recover
    :fn poll-recover-input-function
    :type #{:input}
    :blockable? true}
   {:lifecycle :lifecycle/poll-recover
    :type #{:function}
    :fn poll-recover-input-function
    :blockable? true}
   {:lifecycle :lifecycle/poll-recover
    :type #{:output}
    :fn poll-recover-output
    :blockable? true}
   {:lifecycle :lifecycle/offer-barriers
    :type #{:input :function}
    :fn offer-barriers
    :blockable? true}
   {:lifecycle :lifecycle/offer-barrier-status
    :type #{:input :function :output}
    :fn offer-barrier-status
    :blockable? true}
   {:lifecycle :lifecycle/recover-input 
    :type #{:input}
    :fn recover-input}
   {:lifecycle :lifecycle/recover-state 
    :type #{:windowed}
    :fn recover-state}
   {:lifecycle :lifecycle/unblock-subscribers
    :type #{:input :function :output}
    :fn unblock-subscribers}
   {:lifecycle :lifecycle/next-iteration
    :type #{:input :function :output}
    :fn next-iteration}
   {:lifecycle :lifecycle/input-poll-barriers
    :type #{:input}
    :fn input-poll-barriers}
   {:lifecycle :lifecycle/prepare-barrier-sync
    :type #{:input :function}
    :fn prepare-barrier-sync}
   {:lifecycle :lifecycle/seal-barriers?
    :fn seal-barriers?
    :type #{:output}
    :blockable? false}
   {:lifecycle :lifecycle/seal-barriers
    :fn seal-barriers
    :type #{:output}
    :blockable? true}
   ;; TODO: double check that checkpoint doesn't occur immediately after recovery
   {:lifecycle :lifecycle/checkpoint-input
    :fn checkpoint-input
    :type #{:input}
    :blockable? true}
   {:lifecycle :lifecycle/checkpoint-state
    :fn checkpoint-state
    :type #{:windowed}
    :blockable? true}
   {:lifecycle :lifecycle/checkpoint-output
    :fn checkpoint-output
    :type #{:output}
    :blockable? true}
   {:lifecycle :lifecycle/offer-barriers
    :fn offer-barriers
    :type #{:input :function}
    :blockable? true}
   {:lifecycle :lifecycle/offer-barrier-status
    :type #{:input :function :output}
    :fn offer-barrier-status
    :blockable? true}
   {:lifecycle :lifecycle/unblock-subscribers
    :type #{:input :function :output}
    :fn unblock-subscribers}
   {:lifecycle :lifecycle/before-batch
    :type #{:input :function :output}
    :fn (build-lifecycle-invoke-fn event :lifecycle/before-batch)}
   {:lifecycle :lifecycle/read-batch
    :type #{:input}
    :fn function/read-input-batch}
   {:lifecycle :lifecycle/read-batch
    :type #{:function :output}
    :fn function/read-function-batch}
   {:lifecycle :lifecycle/after-read-batch
    :type #{:input :function :output}
    :fn (build-lifecycle-invoke-fn event :lifecycle/after-read-batch)}
   {:lifecycle :lifecycle/apply-fn
    :type #{:input :function :output}
    :fn apply-fn}
   {:lifecycle :lifecycle/after-apply-fn
    :type #{:input :function :output}
    :fn (build-lifecycle-invoke-fn event :lifecycle/after-apply-fn)}
   {:lifecycle :lifecycle/assign-windows
    :type #{:windowed}
    :fn assign-windows}
   {:lifecycle :lifecycle/prepare-batch
    :type #{:input :function :output}
    :fn prepare-batch}
   {:lifecycle :lifecycle/write-batch
    :type #{:input :function :output}
    :fn write-batch
    :blockable? true}
   {:lifecycle :lifecycle/after-batch
    :type #{:input :function :output}
    :fn (build-lifecycle-invoke-fn event :lifecycle/after-batch)}
   {:lifecycle :lifecycle/offer-heartbeats
    :type #{:input :function :output}
    :fn offer-heartbeats}])

(defn filter-task-lifecycles 
  [{:keys [onyx.core/task-map onyx.core/windows onyx.core/triggers] :as event}]
  (let [task-type (cond-> #{(:onyx/type task-map)}
                    (or (not (empty? windows))
                        (not (empty? triggers)))      
                    (conj :windowed))] 
    (->> (build-lifecycles event)
         (filter (fn [lifecycle]
                   (not-empty (clojure.set/intersection task-type (:type lifecycle)))))
         (vec))))

(defn send-heartbeats! [state]
  (let [messenger (get-messenger state)]
