Async example in clojure-py

gistfile1.clj

(ns async-test
  (:require time)
  (:require urllib))

;; See https://github.com/halgari/clojure-py/issues/135
(defmacro time [expr]
  `(let [start# (time/clock)
         ret# ~expr]
     (prn (str "Elapsed time: " (py/round (* (- (time/clock) start#) 1000) 3) " msecs"))
     ret#))

(defmacro await-async [& body]
  `(py.bytecode/YIELD_VALUE ~@body))

(defn download
  [url]
  (future
    (with-open [x (urllib/urlopen url)]
      (println "Start download of " url)
      (.read x))))

(defmacro let-async
  [[& bindings] & body]
  (let [bindings* (mapcat (fn [[k v]] [k `(await-async ~v)])
                          (partition 2 bindings))]
    `(let ~(vec bindings*)
       ~@body)))

(defn waterfall [x]
  (dotimes [t x]
    (let-async [g (download "http://www.google.com")
                y (download "http://www.yahoo.com")
                b (download "http://www.bind.com")]
      (println "--  Downloaded " (count g) " bytes of Google")
      (println "--  Downloaded " (count y) " bytes of Yahoo")
      (println "--  Downloaded " (count b) " bytes of Bing"))))

;; See https://github.com/halgari/clojure-py/issues/133
(def deref-hack deref)

(defn future-comp [& xs]
  (reify
    IPending
    (isRealized [_]
      (every? #(.isRealized %) xs))
    clojure.lang.ideref/IDeref
    (deref [this]
      ;; Hacky busy loop!
      (loop [_ nil] ; clojure-py chokes on empty recurs. See https://github.com/halgari/clojure-py/issues/132
        (if (.isRealized this)
          (map deref-hack xs)
          (recur nil))))))

(defmacro let-parallel
  [[& bindings] & body]
  (let [keys (take-nth 2 bindings)
        vals (take-nth 2 (next bindings))]
    `(let [xs# (await-async (future-comp ~@vals))
           [~@keys] (do (prn (count xs#)) xs#)]
       ~@body)))

(defn parallel [x]
  (dotimes [t x]
    (let-parallel [g (download "http://www.google.com")
                   y (download "http://www.yahoo.com")
                   b (download "http://www.bind.com")]
      (println "--  Downloaded " (count g) " bytes of Google")
      (println "--  Downloaded " (count y) " bytes of Yahoo")
      (println "--  Downloaded " (count b) " bytes of Bing"))))

(defn run-test [f desc]
  (time
    (do
      (println "Begin" desc)
      (try
        (let [asyncfn (f 4)]
          (loop [val (.next asyncfn)]
            (let [async (.send asyncfn @val)]
              (print "|")  ; Here we're in the control loop
              (recur async))))
        (catch py/StopIteration e nil))
      (println "End" desc)))
  (println))

(run-test waterfall "Waterfall (one by one, in order)")
(run-test parallel "Parallel (all at the same time)")