kafka_streams.clj

(defn kafka-streams-router-builder
  [app
   {routing-topic :routing-topic
    effects-topic :effects-topic
    :as router-opts}]
  (let [builder (StreamsBuilder.)]
    (-> builder
        (.stream routing-topic (Consumed/with
                                (k.serde/string-serde)
                                (k.serde/edn-serde)))
        (.flatMapValues
         (reify ValueMapper
           (apply [_ v]
             ;; (info "routing" v)
             (routing/route-message app v))))
        (.through effects-topic (Produced/with
                                 (k.serde/string-serde)
                                 (k.serde/edn-serde)))
        (.foreach
         (reify ForeachAction
           (apply [_ k v]
             ;; (info "effects" v)
             (effects/apply-effect app v)))))

    builder))