For a channel c, I’ve some code like:

(let [latch (atom (count urls))]
  (doseq [url urls]
    (thread
      (let [resp (http/get url ...)]
        (when (some-condition resp)
          (>!! c (do-something resp))))
      (when (zero? (swap! latch dec))
        (close! c)))))

@xiongtx take a look at async/pipeline-blocking, it maintains ordering (might not be what you want.

But that's the general idea, create a pipeline, flow data through it

Hmm…is pipeline-blocking or pipeline-async more appropriate in this case? I don’t care about the order of the responses on the output chan. Also: - Does pipeline-*’s parallelism have the same caveat r.e. blocking ops as go blocks? https://eli.thegreenplace.net/2017/clojure-concurrency-and-blocking-with-coreasync#blocking-i-o - I’d assume pipeline-* spins up n threads, where n is the degree of parallelism. Then it’d be bad to block any of those n threads, yes? - Does the channel parameter of af in pipeline-async correspond to the to chan of pipeline-async itself? I assume so, but the docs aren’t clear on that.

Is there a a more idiomatic way to "get all values off a channel until the channel is closed" than:

(loop []
  (when-let [v (<!! c)]
    (do-something v)
    (recur)))

You can reduce off the channel