 
   (init-pipeline! config)
 
+  (keys (compile-config config))
+
   (def stream (-> config :streams first))
 
   (defn out-> [all-output]



(defn init-stores [config]
  (let [streams (map :input-topic (filter (where :state = :partitioned) (:streams config)))
        _ (log/info "Initializing kv-store for the following streams: " streams)
        local-store (constantly (thread-local (kv/make-in-memory-kvstore)))]
    (into {} (map (juxt identity local-store) streams))))

          (var-set kv-state new-state'))))))


(defn- compile-input-topic-config [{:keys [input-topic state] :as stream} config]
  (when (or (= state :none) (= state :partitioned))
    {input-topic {:topic-filter (compile-partition-filter stream)
                  :handler (compile-input-topic-handler stream config)}}))

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

(defn- compile-state-topic-config [{:keys [state-topic state input-topic] :as stream}]
  (cond

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

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


(defn- compile-stream-config
  [stream config]
