Async Agents via core.async

gistfile1.clj

(use 'clojure.core.async)

(defprotocol ISendable
  (channel [this]))

(defn async-agent [state]
  (let [c (chan Long/MAX_VALUE) ;; <<-- unbounded buffers are bad, but this matches agents
        a (atom state)]
    (go-loop []
             (when-let [[f args] (<! c)]
               (reset! a (<! (apply f @a args)))
               (recur)))
    (reify
      ISendable
      (channel [this] c)
      clojure.lang.IDeref
      (deref [this] @a))))


(defn send! [a f & args]
  (put! (channel a) [f args]))


(def a (async-agent 0))



(defn async-add [a b]
  (go (+ a b)))

(defn delayed-async-add [a delay b]
  (go (<! (timeout delay))
      (+ a b)))

(send! a async-add 1)

(deref a)

;;; Execute the following code several times, and then notice how the deref of a slowly increments. 

(send! a delayed-async-add 1000 1)