   (fn [output-collector stream partition key message]
     (kv-restore! input-topic message)))
 
+(defn- compile-global-kv-state-handler [{:keys [input-topic state] :as stream}]
+  (fn [output-collector stream partition key message]
+    (global-kv-restore! input-topic message)))
+
 (defn- compile-state-topic-config [{:keys [state-topic state input-topic] :as stream}]
   (cond
 

  "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

      ;; other messages are processed by the normal pipeline, however here
      ;; we need to emit the state messages as well.

      ;; this ugly stuff works because *stores* is thread-local
      (let [kv-state (get *stores* stream)
            state    (var-get kv-state)
            [new-state rich-events] (pipeline state message)
            txlog     (kv/tx-log new-state)
