        (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))))

    (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))))



           (trk/get-metric "events.all.counter"))
         :unknown)))))


;; TODO: remove this after kernel2 migration
(defn events-counter-stats []
  (all-events-counter :current-value))

