    []))
 
 
+
 (defn local-kv-restore
-  "Takes a bunch of tx-log events and restores them in the stream loca-stores"
+  "Takes a bunch of tx-log events and restores them in the stream local-stores"
   ([{:keys [component stream-id events]}]
    (let [kv-store (:local-store (stream-config component stream-id))]
      (var-set kv-store

  (var-set [this new-val] (.set this new-val))
  (var-get [this] (.get this)))



(defrecord SamsaraCore
    [config            ;; normalized configuration currently used
     dispatchers       ;; route dispatchers
     global-stores     ;; global-stores for dimensions
     stores]           ;; stream-local data stores (thread-locals)
    component/Lifecycle


  (start [{done :initialized :as this}]
    (if-not done
      (-> this
          (assoc :global-stores (atom {}))
          (assoc :initialized   true))
      this))


  (stop [{done :initialized :as this}]
    (if done
      (-> this
          (assoc :global-stores nil)
          (assoc :stores        nil)
          (assoc :initialized   false))
      this)))



  "Return the configuration for the given stream"
  [component stream-id]
  (get-in component [:config :streams stream-id]))



(defn component-stream-processor-wrapper
  [component stream-id handler]
  (fn [values]
    (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))))


(defn- tx-log->formatter
  "Return a function which turns the tx-log into a a triplet of

   [topic partition-key json-data]"
  [kvstore-topic]
  (fn [txlog]
    (->> txlog
         (map (fn [[k ver v]]
                [kvstore-topic k (to-json [k ver v])])))))

;; TODO: change txlog format look like events
;; TODO: support dispatch-fn and  format fn
(defn topic-process-wrapper [component stream-id pipeline]
  (let [{:keys [local-store state-topic
                output-topic-partition-fn output-topic ]
         } (stream-config component stream-id)
        output-collector (:output-collector component)
        tx-log->messages (tx-log->formatter state-topic)
        out-dispatch (fn [event] [output-topic
                                 (output-topic-partition-fn event)
                                 (to-json event)])]
    (fn [events]
      (let [state    (var-get local-store)
            [new-state rich-events] (pipeline state events)
            txlog     (kv/tx-log new-state)
            txlog-msg (tx-log->messages txlog)
            output-events (map out-dispatch events)
            all-output (concat txlog-msg output-events)]

        ;; emitting the output
        (@output-collector all-output)

        ;; flushing tx-log
        (let [new-state' (kv/flush-tx-log new-state txlog)]
          (var-set local-store new-state'))))))


(defn load-function-from-name
  ([fqn-fname]
   (if (string? fqn-fname)
     (let [[fns ff] (str/split fqn-fname #"/")]
       (load-function-from-name fns ff))
     fqn-fname))
  ([fn-ns fn-name]
   ;; requiring the namespace
   (require (symbol fn-ns))
   (let [fn-symbol (resolve (symbol fn-ns fn-name))]
     (when-not fn-symbol
       (throw (ex-info (str "function '" fn-ns "/" fn-name "' not found.") {})))
     fn-symbol)))


(defn processor-wrapper
  ([component stream-id]
   (let [{:keys [processor-type processor]} (stream-config component stream-id)]
     (processor-wrapper component stream-id processor-type processor)))

  ([component stream-id processor-type processor]
   (log/debug "Initializing processor [" stream-id "/" processor-type "]:" processor)
   (let [handler (load-function-from-name processor)]
     (case processor-type
       :component-stream-processor
       (component-stream-processor-wrapper component stream-id handler)

       :samsara-factory
       (let [handler' (handler (:config component))]
         (topic-process-wrapper component stream-id handler'))

       :raw
       handler))))



(defn build-routes-for-stream [component stream-id]
  (log/debug "Building routes for:" stream-id)
  (let [{:keys [input-topic state-topic state]} (stream-config component stream-id)]
    (cond

      ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
      ;; build route for partitioned topic with no state
      (= state :none)
      [;; add route to handle topic data
       [(where :topic = input-topic)
        (->> (processor-wrapper component stream-id)
             (standard-stream-wrappers component stream-id))]]


      ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
      ;; build route for partitioned topic with partitioned state
      (= state :partitioned)
      [;; add route to handle kv-store
       [(where :topic = state-topic)
        (->> (processor-wrapper component stream-id
                                :component-stream-processor local-kv-restore)
             (standard-stream-wrappers component stream-id))]
       ;; add route to handle topic data
       [(where :topic = input-topic)
        (->> (processor-wrapper component stream-id)
             (standard-stream-wrappers component stream-id))]]


      ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
      ;; build route for global state
      (= state :global)
      [[(where :topic = input-topic)
        (->> (processor-wrapper component stream-id)
             (standard-stream-wrappers component stream-id))]]


      ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
      ;; build route for CUSTOM managed routes
      (= state :custom)
      [[(where :topic = input-topic)
        (->> (processor-wrapper component stream-id)
             (standard-stream-wrappers component stream-id))]]
      )))


(defn build-routes [component]
  (mapcat (partial build-routes-for-stream component) (:priority (:config component))))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;                                                                            ;;
;;                          ---==| T O O L S |==----                          ;;
;;                                                                            ;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;


(def all-events-counter
  "return a function which is a counter
   when called without argument the counter is increased
   when called with :current-value it returns its current
   value.

      (all-events-counter) ;; increase the counter

      (all-events-counter :current-value)
      ;; => {:count 11}

  "
  (let [tmp-registry (new-registry)
        counter (binding [trk/*registry* tmp-registry]
                  (count-tracker "events.all.counter"))]
    (fn
      ([]
       (counter))
      ([x]
       (case x
         :current-value
         (binding [trk/*registry* tmp-registry]
           (trk/get-metric "events.all.counter"))
         :unknown)))))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;                                                                            ;;
;;                 ---==| * S C R A T C H   A R E A * |==----                 ;;
;;                                                                            ;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;


(comment
