                                                              (Thread/sleep 1000)
                                                              (.println (System/out) (str (:onyx.core/id event) " RATE: " (im/snapshot! rate)))))}))
    :lifecycle/after-batch (fn after-batch [event lifecycle]
-                            (im/update! (:generator/rate event) (count (:onyx.core/batch event)))
+                            (let [cnt (count (:onyx.core/batch event))] 
+                              (swap! total-segments + cnt)
+                              (im/update! (:generator/rate event) cnt))
                             {})})
 
 (def in-calls 

            [onyx.plugin.bench-plugin]
            [onyx.static.logging-configuration :as log-config]
            [onyx.plugin.core-async]
            [interval-metrics.core :as im]
            [onyx.api])
  (:import [onyx.plugin RandomInputPlugin]))


(def scheduler :onyx.job-scheduler/balanced)

(def id (java.util.UUID/randomUUID))

(def config (load-config))

(def env-config (assoc (:env-config config) :onyx/id id))
(def peer-config (assoc (:peer-config config) :onyx/id id))

(def env (onyx.api/start-env env-config))

(def peer-group 
  (onyx.api/start-peer-group peer-config))

(def batch-size 20)
(def batch-timeout 10)

(defn my-inc [{:keys [n] :as segment}]
  (assoc segment :n (inc n)))

(def catalog
  [{:onyx/name :in
    :onyx/plugin :onyx.plugin.bench-plugin/generator
    :onyx/type :input
    :onyx/medium :generator
    :onyx/max-pending 50000
    :onyx/batch-timeout batch-timeout
    :onyx/batch-size batch-size}

   {:onyx/name :inc1
    :onyx/fn :onyx.plugin.bench-plugin-test/my-inc
    :onyx/type :function
    :onyx/batch-timeout batch-timeout
    :onyx/batch-size batch-size}

   {:onyx/name :inc2
    :onyx/fn :onyx.plugin.bench-plugin-test/my-inc
    :onyx/type :function
    :onyx/batch-timeout batch-timeout
    :onyx/batch-size batch-size}

   {:onyx/name :inc3
    :onyx/fn :onyx.plugin.bench-plugin-test/my-inc
    :onyx/type :function
    :onyx/batch-timeout batch-timeout
    :onyx/batch-size batch-size}
