This page is not created by, affiliated with, or supported by Slack Technologies, Inc.
2020-04-30
Channels
- # babashka (46)
- # beginners (234)
- # bristol-clojurians (4)
- # cider (7)
- # clj-kondo (39)
- # cljdoc (8)
- # cljs-dev (10)
- # cljsjs (10)
- # cljsrn (24)
- # clojure (84)
- # clojure-brasil (7)
- # clojure-europe (12)
- # clojure-germany (4)
- # clojure-italy (3)
- # clojure-nl (41)
- # clojure-spec (17)
- # clojure-uk (66)
- # clojurescript (64)
- # conjure (161)
- # cursive (12)
- # data-science (45)
- # datomic (20)
- # devops (11)
- # docker (2)
- # duct (9)
- # events (7)
- # figwheel (1)
- # figwheel-main (20)
- # fulcro (32)
- # graalvm (5)
- # helix (82)
- # jackdaw (9)
- # jobs-discuss (19)
- # kaocha (11)
- # local-first-clojure (1)
- # malli (6)
- # meander (3)
- # nrepl (12)
- # off-topic (2)
- # other-lisps (15)
- # pathom (14)
- # rdf (6)
- # re-frame (8)
- # reactive (1)
- # reagent (5)
- # reitit (4)
- # rum (3)
- # shadow-cljs (77)
- # spacemacs (3)
- # sql (9)
- # test-check (31)
- # tools-deps (13)
- # vim (62)
- # xtdb (18)
Hi everyone,
i have tried the based example of simple topology but i didn't get any results printed
(ns hello.text
(:require [jackdaw.streams :as js]
[jackdaw.client :as jc]
[jackdaw.client.log :as jcl]
[jackdaw.admin :as ja]
[jackdaw.serdes :refer [string-serde]]
[jackdaw.serdes.json :refer [serde]]
[jackdaw.serdes.resolver :as resolver]))
(def kafka-config
{"application.id" "hello
"bootstrap.servers" kafka-url
"default.key.serde" "jackdaw.serdes.EdnSerde"
"default.value.serde" "jackdaw.serdes.EdnSerde"
"cache.max.bytes.buffering" "0"})
;; Serdes tell Kafka how to serialize/deserialize messages
;; We'll just keep them as JSON
(def serdes
{:key-serde (string-serde)
:value-serde (serde)})
;; Each topic needs a config. The important part to note is the :topic-name key.
(def source-topic
(merge {:topic-name "source"
:partition-count 1
:replication-factor 1
:topic-config {}}
serdes))
(def dest-topic
(merge {:topic-name "dest"
:partition-count 1
:replication-factor 1
:topic-config {}}
serdes))
(defn simple-topology [builder]
(-> (js/kstream builder source-topic)
(js/for-each! (fn [[key msg]]
(prn msg)
))
))
(defn view-messages [topic]
"View the messages on the given topic"
(with-open [consumer (jc/subscribed-consumer
(assoc kafka-config "group.id" (str (java.util.UUID/randomUUID)))
[topic])]
(jc/seek-to-beginning-eager consumer)
(->> (jcl/log-until-inactivity consumer 100)
(map :value)
doall)))
(defn start! []
"Starts the simple topology"
(let [builder (js/streams-builder)]
(simple-topology builder)
(doto (js/kafka-streams builder kafka-config)
(js/start))))
(defn stop! [kafka-streams-app]
"Stops the given KafkaStreams application"
(js/close kafka-streams-app))
(def hello (start!))
i'm executing that in cider emacs
but i don't get any message printed
is there something i did wrong ?
if i run that separately
(def builder (js/streams-builder))
(simple-topology builder) ;; that gives nil, is it impure function which start something in the background ?
topologies start threads
I don't see anything in your example that writes messages the topology would consume?