 
 (defn init-pipeline! [config]
   (log/info "Initialize processors...")
-  (alter-var-root #'*config*       (constantly config))
-  (alter-var-root #'*store*        (constantly (init-stores config)))
-  (alter-var-root #'*dispatchers*  (constantly (compile-config config))))
+  (alter-var-root #'*config*        (constantly config))
+  (alter-var-root #'*stores*        (constantly (init-stores config)))
+  (alter-var-root #'*global-stores* (constantly (atom {})))
+  (alter-var-root #'*dispatchers*   (constantly (compile-config config))))
 
 
 

  "Takes an event in json format which contain an Tx-Log entry and
   it pushes the change to the running *kv-store*"
  [stream ^String event]
  ;; this ugly stuff works because *store* is thread-local
  ;; TODO: make sure initialization adds all the streams
  (printf "STATE[%s] : %s\n" stream event)
  (let [kv-store (get *store* stream)]
    (var-set kv-store
             (->> event
                  from-json
                  vector
                  (kv/restore (var-get kv-store))))))



(defn process-dispatch


    (= state :global)
    {input-topic {:topic-filter (compile-partition-filter stream)
                  :handler (compile-global-kv-state-handler stream)}}
    ))


