 
 (defn to-json-format-wrapper
   "a wrapper which convert the :message from a clojure data structure
-   into ajson formatted string."
+   into a json formatted string."
   [handler]
   (fn [values]
     (handler (map #(update % :message to-json) values))))

(defn- stream-defaults
  "Apply defaults for a stream configuration"
  [{:keys [input-topic state-topic state processor processor-type] :as stream}]
  (as-> stream it
    ;; add default state-topic
    (if (and (= :partitioned state) (not state-topic))
      (-> it
          (assoc :state-topic (str input-topic "-kv"))
          (assoc :local-store (thread-local (kv/make-in-memory-kvstore))))
      it)
    ;; when state is global use global-kv-restore!
    (if (and (= :global state) (not processor))
      (-> it
          (assoc :processor #'global-kv-restore)
          (assoc :processor-type :component-stream-processor))
      it)
    ;; when state is global use global-kv-restore!
    (if (and (= :global state) (not processor-type))
      (assoc it :processor-type :component-stream-processor)
      it)

    ;; merge defaults
    (merge DEFAULT-STREAM-CFG it)))




;; TODO: tracking needs to be reviewed
;; TODO: display of events
;; TODO: per message vs per batch error handling
;; TODO: handling of events which not match any route
;;


