Clojurians Log v2

To return via reduced. I just return reduced as a value? Thanks a lot guys!!

With take-while the predicate just needs to return false, this will wrap the result from the last valid input in a reduced (https://github.com/clojure/clojure/blob/clojure-1.12.4/src/clj/clojure/core.clj#L2919).

so most likely, when an ap emits (m/amb (reduced nil)), then the consumer will stop as well.

Calling reduced only makes sense in the context of a reducing function, see https://clojure.org/reference/transducers. Calling it within m/ap just emits a value that happens to be a clojure.java.Reduced instance (https://github.com/clojure/clojure/blob/clojure-1.12.4/src/clj/clojure/core.clj#L2853), and unless m/ap is within a reducing function (I cannot think of why one would want that or if it's even possible or making sense) it will just be another emitted value like any other.

the task that will consume a flow is a reducer. so I think it should work to return (reduced nil) or (m/ap (reduced nil)). I will do some tests.

I'm having a hard time imagining how a missionary task would work in a reducing function, sounds like a sync/async mismatch, but as you say the proof is in the pudding!

You were right. The emitted value of a flow is then just (reduced val).

(ns demo.flows.clock
  (:require
   [missionary.core :as m]))

(def >clock    ;; A shared process emitting increasing numbers every second.
  (m/stream
   (m/ap
    (println "creating clock!")
    (loop [i 0]
      (m/amb
       (m/? (m/sleep 1000 i))
       ;(println "i: " i)
       (recur (inc i)))))))



(def stopping-clock
  (m/ap
   (println "creating clock!")
   (let [v (m/?> >clock)]
     (if (> (rand 100) 90.0)
       (do
         (println "stopping clock.")
         (reduced nil))
       v))))

(m/? (m/reduce println stopping-clock))

creating clock!

nil 0
nil 1
nil 2
nil 3
nil 4
nil 5
nil 6
nil 7
stopping clock.
nil #reduced[nil]
nil 9
nil 10
nil 11

(defn print-while-not-reduced [s v]
  (println "value: " v)
  v)

(m/? (m/reduce print-while-not-reduced nil stopping-clock))

creating clock!
value:  0
value:  1
stopping clock.
value:  #reduced[nil]

in that case you can just move the stopping logic to the reducing function you pass to m/reduce (i.e. you don't need the stopping-clock flow), or if you want a flow with that logic, it is better to use eduction, and not depend on the consumer having to find the "terminating" value

At this stage it really would help to know the actual source of your flow. Something you need to create? Some stream provided to you? ...

well, in the reducing function I can only use synchronous code. so if the exit condition needs a m/sp then I need to emit some kind of value that the reducer will know needs to terminate.

i think this (m/>? (m/eduction (take-while your-pred) flow)) should do the trick, when your-pred returns false you stop getting values from the flow, and the ap flow stop producing them

m/ap terminates naturally when the underlying flow passed to m/?> terminates naturally. When passed an m/seed flow it terminates when the collection is entirely processed. The other mechanism to terminate naturally that I'm aware of would require a source that explicitly supports a "complete" callback that calls the terminator function (2nd arg to a custom flow function, see https://github.com/leonoel/flow). Not sure the solution of using m/eduction as suggested will do a graceful termination, as the doc says Early termination by the transducing stage (via reduced or throwing) cancels upstream flow, so there may be a risk of cancellation before the last value is processed (I think).

I guess the real difference between calling the terminate fn given to a custom flow function, and the m/eduction way, is where you want to trigger the termination: in the underlying custom flow code given to m/?> or in the consumer of that underlying flow (`m/eduction` within the m/ap).

Or you can just create a flow by hand, it's just a function

https://nextjournal.com/N-litened/missionary-for-dummies#flows

A warning: I think some of the info at that link is Missionary internals, and that some of it is out of date. That was the conclusion I came to when looking at it a few months ago, but I don’t recall details. (Apologies if this is misinformation.) @leonoel It would be great if you could say whether that’s true.