Hey,
I'm doing some core.async tests and want to create a basic pub/sub model.
Messages are >! on one channel and >! to many others.
(deftest ^:wip async-test2
(let [subscribers (atom [])
events (chan 100)]
(go (loop []
(when-let [ev (<! events)]
(doseq [c @subscribers]
(alt!
[[c ev]] :sent
:default nil ;; could "force" unsubscribe c?
))
(recur))))
(let [s1 (chan 1)
s2 (chan 100)
r1 (atom [])
r2 (atom [])]
(swap! subscribers conj s1 s2)
;; simulated slow reader
(go (loop []
(when-let [ev (<! s1)]
(swap! r1 conj ev)
(<! (timeout 10))
(recur))))
;; good reader
(go (loop []
(when-let [ev (<! s2)]
(swap! r2 conj ev)
(recur))))
(<!! (go (loop [i 0]
(when (< i 100)
(>! events i)
(recur (inc i))))))
(close! events)
(Thread/sleep 25)
(pprint @r1)
(pprint @r2))
))
In this example the s1 subscriber will loose almost all messages since he
cannot keep up (and buffer is too small). I choose to alt!/:default to drop
messages, since I don't want any subscriber to block others. How do you
guys deal with slow-readers?
I don't really have a specific problem but I wonder if there are any plans
for some built-in pub/sub mechanisms for core.async. Seems like a very
common pattern.
Anyways, core.async is nice!
Cheers,
/thomas
--
--
You received this message because you are subscribed to the Google
Groups "Clojure" group.
To post to this group, send email to [email protected]
Note that posts from new members are moderated - please be patient with your
first post.
To unsubscribe from this group, send email to
[email protected]
For more options, visit this group at
http://groups.google.com/group/clojure?hl=en
---
You received this message because you are subscribed to the Google Groups
"Clojure" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
For more options, visit https://groups.google.com/groups/opt_out.
