2025-01-16 missionary | Clojure Slack Archive

missionary 2025-01-16

eduardomrb 2025-01-16T22:30:35.313719Z

Hi, I've been having lots of fun playing with missionary, it's really cool! So I'm trying to write this sort of background job that is supposed to run indefinitely. It consists in two parts: get a list of resources from an external API in a loop but only emit the distinct items, then create other flows (one for each item) that will continuously call another endpoint that retrieves more information about each of those resources until it reaches a certain status. I managed to get the first part working with:

;; get-service-deployments returns a list
(defn <service-deployments [service]
  (m/ap
    (let [service-deployments (m/?> (m/seed (repeatedly #(get-service-deployments service))))]
      (m/? (m/sleep 1000))
      service-deployments)))
      
(def main
  (let [<sds (->> (<service-deployments "service-name")
               (m/reductions {} nil)
               (m/eduction
                 cat
                 (distinct)))]
    (m/reduce (fn [_ sd] (prn sd)) nil <sds)))

However, I can't figure out a way to do the second part. I'd really appreciate any tips. Thank you!

leonoel 2025-01-17T08:05:34.746249Z

You can nest ?> in an ap block. In this case you probably want to allow nested flows to run concurrently, therefore you will call the outer ?> with a concurrency limit. You can use m/eduction with take-while to stop running a flow when a condition is met.

(defn poll [task delay]
  (m/ap (m/? (m/sleep delay (m/? (m/?> (m/seed (repeat task))))))))

(def concurrency 10)

(def main
  (m/reduce (fn [_ x] (prn x)) nil
    (m/ap
      (let [sd (m/?> concurrency
                 (->> (poll (m/via m/blk (get-service-deployments "service-name")) 1000)
                   (m/eduction cat (distinct))))]
        (m/?> (->> (poll (m/via m/blk (another-endpoint sd)) 1000)
                (m/eduction (take-while some-condition))))))))

leonoel 2025-01-17T08:08:24.787039Z

In your example, get-service-deployments is certainly a blocking call so you have to make it explicitly async with m/via

eduardomrb 2025-01-17T12:45:37.677589Z

that makes sense now, thank you

eduardomrb 2025-01-17T12:49:47.498649Z

just one more question, for some reason I'm not getting the results of the nested flow prnted by the m/reduce, do I need to do something to emit those values?

eduardomrb 2025-01-17T13:12:06.061429Z

nvm, I've made a mistake on the take-while pred, the prn works as expected

Clojurians Log v2

missionary 2025-01-16