This page is not created by, affiliated with, or supported by Slack Technologies, Inc.
2020-04-29
Channels
- # announcements (35)
- # aws (40)
- # babashka (10)
- # beginners (119)
- # calva (25)
- # cider (13)
- # clj-kondo (15)
- # cljsrn (23)
- # clojure (205)
- # clojure-dev (3)
- # clojure-europe (15)
- # clojure-germany (3)
- # clojure-italy (3)
- # clojure-nl (2)
- # clojure-uk (58)
- # clojurescript (193)
- # community-development (2)
- # conjure (147)
- # core-async (49)
- # cursive (47)
- # datomic (27)
- # duct (1)
- # fulcro (19)
- # graalvm (3)
- # graphql (1)
- # helix (3)
- # hoplon (11)
- # jackdaw (1)
- # joker (1)
- # juxt (5)
- # kaocha (1)
- # keechma (3)
- # lambdaisland (6)
- # local-first-clojure (27)
- # malli (5)
- # off-topic (41)
- # rdf (27)
- # re-frame (7)
- # reagent (15)
- # reitit (5)
- # rum (11)
- # shadow-cljs (157)
- # spacemacs (18)
- # sql (4)
- # xtdb (8)
I have general question, about consuming messages from Kafka topic, is it better to collect all messages into async channel and handle it by pure methods which take channel and process it and return new output channel and so on or what is the best approach to model that ?
I responded already in the other channel, but more specifically about core.async here - what specifically would core.async give you when processing kafka- what feature are you looking for?
I'd say use transducers instead of "pure" methods which take channel. Attach transducers to channels or pipelines.
also you'll need to decouple the interaction between kafka and core.async, either can introduce lags that break the other
hello, I just hit an issue while trying to use loop + try/catch, this is a code that breaks:
(defn x [some-chan]
(go
(loop []
(try
(when (<! some-chan)
(recur))
(catch :default e
(recur))))))
(reduced to make a point), but this code doesn't compile
it does the error
------ ERROR -------------------------------------------------------------------
File: /Users/wilkerlucio/Development/relemma/src/cljs/relemma/shared/logic/feature_flags.cljs:57:3
--------------------------------------------------------------------------------
54 | unlisten-cb))
55 |
56 | (defn x [some-chan]
57 | (go
---------^----------------------------------------------------------------------
Encountered error when macroexpanding cljs.core/cond.
IllegalArgumentException: No implementation of method: :emit-instruction of protocol: #'cljs.core.async.impl.ioc-macros/IEmittableInstruction found for class: cljs.core.async.impl.ioc_macros.Jmp
clojure.core/-cache-protocol-fn (core_deftype.clj:583)
clojure.core/-cache-protocol-fn (core_deftype.clj:575)
is there a way around it?
yeah, but then when the channel is closed this would loop like crazy, right?
that behavior would be shared by the original
not really, the original has a when
for no errors, and no recur in that case
oh, you guys are right, because of the reduced code
@wilkerlucio I don't see a when, and both the try and the catch recur to the same place
but on the original I had, there was a when aroudn
snippet updated
err, they would recur to the same place if recur across try was valid
which it isn't
got it working with this:
(defn x [some-chan]
(go
(let [continue* (atom true)]
(loop []
(try
(if (<! some-chan)
(do-something)
(reset! continue* false))
(catch :default e
(handle-error)))
(if @continue*
(recur))))))
for that sort of thing, I like promise
or delay
, which convey the extra semantic that it's a one way switch
in my case its a streaming thing
agreed for single hit a promise would be great
right, but nothing ever changes the atom to true - it's either not-done, or done
it does, (snippet updated)
when channel comes with nil
, it stops
I still don't see a case where it's assigned to true - you create a new one
not sure if I get it, the point is just that one message can come as an error, in this case I want to report and wait for the next message
the continue starts with true, but if the channel closes its set to false, and recur stops
the point is that you have created a one-way-switch - once created it only changes once, from true to false
yeah, that's when the channel gets closed, so nothing more to get from it
after closing there is nothing else to do
I'm not saying the atom is incorrect, I'm saying delay / promise are more specialized, and they are less error prone and more clear to a reader for that reason
it's a style suggestion not a bug fix
principle of least power
but how I could use that since this is a stream channel?
I expect multiple messages from it
it literally replaces the atom, you create it in the same place, instead of using @
as the check, you use realized?
, and instead of reset! you use deliver
or force
ah, gotcha, also I'm on CLJS land, you still thing would be better to use something else than atom there?
(let [done? (promise)] (if (not (realized? done)) (try ... (catch ... (deliver done? :error-state)))
oh, I'm forgetting cljs might not have those constructs
cljs appears to have delay
but not promise
in the random cljs repl I checked
then you could use (done (delay true))
then later (force done)
@wilkerlucio recur doesn't work across try boundaries, whether core.async or not