Personally I do something like this:

(defn fetch-something [...])

(defn ?fetch-something [...]
  (m/via m/blk (fetch-something ...)))

// The caller is responsible to use `m/?`
(m/? (?fecth-something ...))

I like the ? prefix, do you use it to indicate a task?

do you have a convention for a flow as well like <? ?

Yes, the ? prefix indicate that the function return a task. For a flow, I use <.

batch / batching / partition / partition-all / manifold.stream/bactch strean/batch s/batch (if someone is looking)

The one thing I am missing is a timeout (take 2 but if the second value takes a while then go with just one)

saw that one day, but didn't try https://clojurians.slack.com/archives/CL85MBPEF/p1707239083211269

I used the wrong term it's not a timeout I'm talking about but a max latency, so if there is a partial batch result but the max-latency is passed since the previous batch, the partial batch is emitted

you can use a modified version of the debounce operator (sorry on mobile can’t check)

(import '(missionary Cancelled))

(defn debounce [dur >in]
  (m/ap
    (let [batch (atom [])
x (m/?< >in)]
      (try (m/? (m/sleep dur x))
           (catch 
             (let [b (swap! batch conj x)
(if (= (count b) 42)
  (do (reset! batch []) (amb b))
             (m/amb))))))

@U03K8V573EC I'm struggling to find a purely functional solution to this problem. Could you share some inspiration from other languages/libraries ?

manifold.stream/batch does that but I doubt it's purely functional

The requirement is a little bit tricky. Does the timeout start at the first value of a partition or after the last seen value from the flow? If we do 3 elements partition and we had value1 2 seconds ago and value2 1 second ago and the timeout is 2 seconds, do we emit a pack or do we still have 1 second to wait after value2 ? Assuming the later case, I have implemented something with delayed values and custom transducers. Every value is emitted with a delayed twin then a custom transducer discard a delayed twin if a new value was in between but if a value and its twin are successive a separator mark is propagated. At last, a custom partition-all transducer can make pack of n values and force partitioning when separation mark is found.

I think this one is close

(defn batch [max-size max-delay input]
  (m/ap (let [[_ input] (m/?> (m/group-by {} input))]
          (m/? (m/reduce conj
                 (m/eduction (take-while (complement #{::timeout})) (take max-size)
                   (m/ap (m/amb= (m/?> input) (m/? (m/sleep max-delay ::timeout))))))))))

what is the purpose of m/group-by in the impl? Why "close", what's missing?

without group-by, the input flow would be cancelled/restarted between successive batches

Can you explain a little bit in more detail, why the group-by is needed? Why is would the flow restarted/cancelled without it?

"if or none" makes sense to me. I came across this pattern several times, the name I have in mind is where (and where-let,`where-some`) because • it's about filtering, like SQL where • it's the same kind of word as when, which has the same siblings with the same syntax

I like where, ifn makes me think of https://clojure.github.io/clojure/javadoc/clojure/lang/IFn.html

@U05P1LZ8EQH would be interesting to run some benchmarks for the different solutions (transducers vs loop recur vs ap with imperative state), I realize I've been using that pattern with transducers without thinking about alternatives, I wonder which one offers the best perfs, for dedupe for instance

For loop/recur using my "transmuter" I wouldn't be surprised if it was a bit slower because it's adding an entire extra layer where values get bounced through an rdv, etc. But just speaking for myself I'm not doing anything CPU bound, so it doesn't really matter to me.