 (ns samsara-core.kernel2
   (:refer-clojure :exclude [var-get var-set])
+  (:require [samsara.trackit :refer [track-time track-rate new-registry
+                                     count-tracker distribution-tracker
+                                     track-pass-count] :as trk])
   (:require [clojure.string :as str]
             [com.stuartsierra.component :as component]
             [moebius
-             [core :refer [where inject-if]]
+             [core :refer [inject-if where]]
              [kv :as kv]]
-            [samsara.utils :refer [from-json to-json]]
-            [taoensso.timbre :as log])
-  (:require [samsara.trackit :refer [track-time track-rate new-registry
-                                     count-tracker distribution-tracker
-                                     track-pass-count] :as trk]))
-
+            [samsara.utils :refer [from-json invariant to-json]]
+            [taoensso.timbre :as log]))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;                                                                            ;;

    (handler {:component component :stream-id stream-id :events values})))


(defn- standard-stream-wrappers
  [component stream-id handler]
  (let [{:keys [format input-partitions]} (stream-config component stream-id)]
    (->> handler
         (message-only-wrapper)
         (format-wrapper format)
         (partition-filter-wrapper input-partitions))))



  (let [cmp (component/start (make-samsara-core config))]
    (alter-var-root #'*pipelines*
                    (constantly
                     (fn
                       ([] cmp)
                       ([output-collector stream partition key message]
                        (reset! (:output-collector cmp) output-collector)
                        (process (:routes cmp) [{:topic stream
                                                 :partition partition
                                                 :key key
                                                 :message message}])))))
    cmp))



;; TODO: this should use the stream-id instead
(def output-collector-tracker
  (memoize
   (fn [job-name stream]
     (let [prefix (str "pipeline." job-name "." stream)
           total-size-out(count-tracker (str prefix ".out.total-size"))
           dist-size-out (distribution-tracker (str prefix ".out.size"))

           size-out-tracker (invariant #(let [sz (count (last %))]
                                          (total-size-out sz)
                                          (dist-size-out sz)))]
       (fn [output-collector]
         (fn [outputs]
           (output-collector (map size-out-tracker outputs))))))))



(defn process-dispatch
