Clojure core.async pipeline example

pipeline.clj

(require '[clojure.core.async :as async]
         '[clj-http.client :as client]
         '[clojure.data.json :as json])

(def concurrency 5)

(let [in (async/chan)
      out (async/chan)
      request-handler (fn [url out*]
                          (async/go
                            (println "Making request:" url)
                            (let [response (client/get url)
                                  body (json/read-str (:body response))]
                              (doseq [repo (body "items")]
                                (async/>! out (repo "clone_url"))))
                            ; Finally close the channel to signal finished processing
                            (async/close! out*)))]
  
  ; Process `in` messages concurrently
  (async/pipeline-async concurrency out request-handler in)

  ; Push URLs to process
  (async/go 
    (doseq [url (for [page (range 10)] (str "https://api.github.com/search/repositories?q=language:clojure&page=" 
                                            (inc page)))]
      (async/>! in url)))
  
  ; Print results of processing
  (async/go-loop []
    (println (async/<! out))
    (recur)))

; `in` can be backed by a redis queue
(comment
  (async/go-loop []
    (if-let [message (pop-redis-queue)]
      (async/>! in message)
      ; Sleep if no messages available
      (async/<! (async/timeout 1000)))
    (recur)))