Agents in Clojure, Part II

01-validation.clj

(def read-agent (agent "" :validator string?))

02-error-handler.clj

(defn read-agent-error-handler [agnt, exception]
  (println "Whoops! " agnt " had a problem: " exception))

(def read-agent (agent "" :error-handler read-agent-error-handler))

03-set-options.clj

(set-validator! read-agent string?)

(set-error-handler! read-agent read-agent-error-handler)

04-callbacks.clj

(defn read-watch [key agnt old-value new-value]
  (println "File has been read!")
  (println (str "Watch key: " key))
  (println (str "Agent: " agnt))
  (println (str "Agent's old value: " old-value))
  (println (str "Agent's new value: " new-value)))

(add-watch read-agent "reader-01" read-watch)

05-agent.clj

(defn read-agent-error-handler [agnt, exception]
  (println "Whoops! " agnt " had a problem: " exception))

(def read-agent
  (agent
    "zero bytes"
    :validator string?
    :error-handler read-agent-error-handler))

(defn big-read [old-value seconds]
  "Pretent to read a really big file"
  (time (Thread/sleep (* seconds 1000)))
  "<contents of big file>")

(defn read-watch [key agnt old-value new-value]
  (println "File has been read!")
  (println (str "New file data is: " new-value))
  (println ""))

(add-watch read-agent "reader-01" read-watch)

06-repl.txt

$ start-clojure
Clojure 1.4.0
user=> (ns async-2)
nil
async-2=> (load-file "05-agent.clj")
#<Agent@2598c6f3: "zero bytes">
async-2=>
async-2=> (send-off read-agent big-read 20)
#<Agent@2598c6f3: "zero bytes">
async-2=>
async-2=> ; let's do something else while we're waiting
async-2=> ; maybe some simple math?
async-2=> (+ 1 1)
2
async-2=>

07-repl.txt

async-2=> "Elapsed time: 10000.776 msecs"
File has been read!
New file data is: <contents of big file>

08-relay-ring.clj

(defn relay [x i]
  (when (:next x)
    (send (:next x) relay i))
  (when (and (zero? i) (:report-queue x))
    (.put (:report-queue x) i))
  x)

(defn run [action-count agent-count]
  (let [q (new java.util.concurrent.SynchronousQueue)
        hd (reduce (fn [next _] (agent {:next next}))
                   (agent {:report-queue q})
                   (range (dec action-count)))]
    (doseq [i (reverse (range agent-count))]
      (send hd relay i))
    (.take q)))

09-repl.txt

async-2=> (time (run 1000 1000))
"Elapsed time: 1003.892 msecs"
0