        (component-stream-processor-wrapper component stream-id handler)
 
        :samsara-factory
-       (let [handler' (handler (:config component))]
-         (topic-process-wrapper component stream-id handler'))
+       (let [handler' (handler (:config component))
+             pipeline (pipeline-stats-wrapper component stream-id handler')]
+         (topic-process-wrapper component stream-id pipeline))
 
        :raw
        handler))))

(ns samsara-core.kernel2
  (:refer-clojure :exclude [var-get var-set])
  (:require [clojure.string :as str]
            [com.stuartsierra.component :as component]
            [moebius
             [core :refer [where inject-if]]
             [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]))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;                                                                            ;;

   is for the kv-store"
  [output-collector stream partition key message]

  (let [collector-tracker (output-collector-tracker
                           (-> (*pipelines*) :config :job :job-name) stream)]

    (all-events-counter)
    (try

      (*pipelines* (collector-tracker output-collector)
                   stream partition key message)

      ;;
      ;; ERROR Handling
      ;;
      (catch Exception x
        (log/warn x "Error processing message from [" stream "]:" message)
        (track-rate (str "pipeline." (-> (*pipelines*) :config :job :job-name)
                         "." stream ".errors"))
        (output-collector [[(str stream "-errors") key message]])))))
