There is no built-in operator to do that, because clojure is already close to optimal here. However you should use sp instead of via to optimize thread usage

I see.. i havent dig what thread used in e.g. (task s f), afaik missionary dont allocate thread pool (where as core.async does) and assume evaluating task will be run by caller thread, is it correct?

Yes, missionary doesn't implicitly run anything on a thread pool like core.async. However, sleep is backed by a scheduler thread, and via is backed by a user-provided thread pool. All tasks are async, so a task process is spawned by the thread requesting it, but it may be completed by another thread. Consequently the following may print two different lines :

(m/sp
  (println "before-sleep" (.getName (Thread/currentThread)))
  (m/? (m/sleep 5000))
  (println "after-sleep" (.getName (Thread/currentThread))))

So the continuatiom will run by the latest thread that got unparked/activated by the "parking" mechanism, e.g. in above example the second println run by sleep scheduler thread ?

yes

Ok thanks, i got more enlightment now 🙏

(defn multiply
  ;; TODO docstring and use n, generally how to fan one item in missionary?
  [_n >x]
  (m/ap
   (let [x (m/?= >x)]
     (m/amb> (assoc x :n 1 :x (rand))
             (assoc x :n 2 :x (* (rand) (rand)))
             (assoc x :n 3 :x (- (rand)))
             (assoc x :n 4 :x (- (rand)))
             (assoc x :n 5 :x (rand))))))

I dont understand the question

In the snippet I’m creating 5 “events” that are going onto the flow. If I had n instead of a fixed number, how would I programmatically put n “events” on the flow?

loop/recur i think

This might help:

(defn chan->flow
  "Produces a discreet flow from a core.async `channel`"
  [channel]
  (m/ap                                                     ; returns a discreet flow
    (loop []
      (if-some [x (m/? (chan-read channel))]                ; read one value from `channel`, waiting until `channel` produces it
        ;; We succesfully read a non-nil value, we use `m/amb` with two
        ;; branches. m/amb will fork the current process (ap) and do two things
        ;; sequencially, in two branches:
        ;; - return x, meaning `loop` ends and return x, ap will produce x
        ;; - recur to read the next value from chan
        (m/amb x (recur))
        ;; `channel` producing `nil` means it's been closed. We want to
        ;; terminate this flow without producing any value (not even nil), we
        ;; use (m/amb) which produces nothing and terminates immediately. The
        ;; parent m/ap block has nothing to produce anymore and will also
        ;; terminate.
        (m/amb)))))

here it is working with a test if helpful

the comments you have in there are super helpful!

thanks I’ll give this a try

if you don't know the number of branches statically you should use (m/?> (m/seed (range n))), look at what amb> desugars to for inspiration

amb> in an explicit loop also works

https://clojurians.slack.com/archives/CL85MBPEF/p1651045172044839

As newcomer i like 1, but i think later in the future 2 will be more idiomatic, like e.g. i tend to use fn like partial or juxt on more occasion than i was previously 🙂