A Simple Async Task in Clojure

Question

Purpose of Code: To simulate scraping the web and updating a db, by, presently, adding randomly generated numbers to a "database".

Have I wired up the followin async code properly? Is there a simpler, more composable way to pass objects around? Or some more idiomatic way?

I have added comments to self-explanatory code to hopefully make it a titch easier on reviewers.

Note: the following mocks what will become a web scraper. So just be aware that the "db" isn't a real db, and "fetching urls" refers to a mock up that will eventually get swapped out, but I hope you see the semantics I intend to sub in.

(ns clj-scraper.core
  (:require [clojure.core.async :as a :refer [>! <! >!! <!! go go-loop chan
                                              buffer close! thread alts!
                                              alts!! timeout
                                              put! take!
                                              pipe pipeline pipeline-async pipeline-blocking]]))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Database - a mockup of what could be a db

(def test-db (agent {:1 {:url  "abc"
                         :vals []}
                     :2 {:url  "bac"
                         :vals []}
                     :3 {:url  "pg"
                         :vals []}))

; append to a path in the test-db
(defn test-append [path x] (send test-db update-in path (partial cons x)))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Load from DB

; objects start their journey on this channel
(def db-objs> (chan 1))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Fetch URLs

; test-data mocks fetching a url by returning the string of a random int
(defn test-data [_] (str (rand-int 100)))

(def pages> (chan 1))
(pipeline 4
          pages>
          ; add the fetched page to the object getting passed around
          (map (fn [{:keys [url] :as obj}] 
                 (conj obj {:page (test-data url)})))
          db-objs>)


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; CSS Select Page

; this mock up simply converts a String -> Int
(defn test-select [x] (Integer. x))

(def selections> (chan 1))
(pipeline 4
          selections>
          ; add the converted String to the object getting passed around
          (map (fn [{:keys [page] :as obj}]
                 (conj obj {:scraped (test-select page)})))
          pages>)


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Update DB

; add the "scraped value" to the "db"
(go-loop []
  (let [{:keys [id scraped]} (<! selections>)]
    (test-append [id :vals] scraped))
  (recur))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Run Things in Debugging

(doseq [[id data] @test-db] (go (>! db-objs> (conj data {:id id}))))
@test-db ; won't show expected results until async ops are done

Answer 1

Commenting on this section:

; test-data mocks fetching a url by returning the string of a random int
(defn test-data [_] (str (rand-int 100)))

(def pages> (chan 1))
(pipeline 4
          pages>
          ; add the fetched page to the object getting passed around
          (map (fn [{:keys [url] :as obj}] 
                 (conj obj {:page (test-data url)})))
          db-objs>)

In real-life, test-data will either do blocking I/O or be asynchronous. Therefore, you should use pipeline-blocking or pipeline-async instead, otherwise you'll exhaust the "go thread pool".

Commenting on this part:

(go-loop []
  (let [{:keys [id scraped]} (<! selections>)]
    (test-append [id :vals] scraped))
  (recur))

You should account for the fact that the selections> channel may close, in which case the object will be nil and you should stop the loop.

(go-loop []
  (when-let [{:keys [id scraped]} (<! selections>)]
    (test-append [id :vals] scraped)
    (recur)))

Commenting on this part:

(doseq [[id data] @test-db] (go (>! db-objs> (conj data {:id id}))))

Because go is asynchronous, this consists of putting all the objects in test-db at once into the processing pipeline, which potentially consumes a lot of memory. Core.async channels are designed to give you back-pressure, you should use it.

How about instead:

(go (doseq [[id data] @test-db] (>! db-objs> (conj data {:id id}))))

Which you can express at a higher-level using onto-chan

(onto-chan db-objs (map (fn [data] (conj data {:id id})) @test-db) false)