Concurrent prime sieve using core.async in clojure

csp-prime-sieve.clj

;;; Concurrent prime sieve using core.async in clojure

(ns primesieve.core
  (:require [clojure.core.async :as async :refer [go-loop chan >! <! <!! close! filter< to-chan]]))

(def bufsize 512)

(defn gen [upper-limit]
  "Returns channel, puts numbers from 2 to upper-limit into channel"
  (to-chan (range 2 upper-limit)))

(defn sieve-filter [in prime]
  "Returns output channel, reads from input channel
   filtering out all numbers divisible by prime"
  (filter< #(pos? (rem % prime)) in bufsize))

(defn sieve-builder [in]
  "Returns output channel, daisy-chains filter processes for every new prime
   received on input channel."
  (let [out (chan bufsize)]
    (go-loop [in in]
      (if-let [prime (<! in)]
        (do (>! out prime)
            (recur (sieve-filter in prime)))
        (close! out)))
    out))

(defn collect [in]
  "Returns output channel, returns vec containing all values from input chan."
  (<!! (async/into [] in)))

(defn primes [upper-limit]
  (-> (gen upper-limit)
      (sieve-builder)
      (collect)))