async.clj

(ns rbl.utils.async
  (:require [clojure.core.async :refer :all]))


(defmacro sleep
  ([ms & stop-channels]
     (let [to (list 'timeout ms)]
       `(alts! ~(into [to] stop-channels))))
  ([ms]
     `(<! (timeout ~ms))))

(defn ref->chan
  "Watches the given reference (atom, ref, agent) and publishes the new values to a channel which is returns"
  ([reference]
     (ref->chan reference (chan)))
  ([reference ref-chan]
     (add-watch reference nil (fn [_ref _key _old new-value] (go (>! ref-chan new-value))))
     ref-chan))

(defmacro gox
  "Like go, but if body raises an exception, then the exception, rather than nil, is put on the returned channel."
  [& body]
  `(go (try ~@body (catch Exception e# e#))))

(defmacro threadx
  "Like thread, but if body raises an exception, then the exception, rather than nil, is put on the returned channel."
  [& body]
  `(thread (try ~@body (catch Exception e# e#))))

(defmacro <!gox
  "Takes the return value of the go body as if by <!, or throws any exception thrown by the body. Must be called inside a (go ...) block."
  [& body]
  `(let [resp# (<! (gox ~@body))]
     (if (instance? Exception resp#)
       (throw resp#)
       resp#)))

(defmacro <!!gox
  "Takes the return value of the go body as if by <!!, or throws any exception thrown by the body."
  [& body]
  `(let [resp# (<!! (gox ~@body))]
     (if (instance? Exception resp#)
       (throw resp#)
       resp#)))

(defmacro <!threadx
  "Takes the return value of the thread body as if by <!, or throws any exception thrown by the body. Must be called inside a (go ...) block."
  [& body]
  `(let [resp# (<! (threadx ~@body))]
     (if (instance? Exception resp#)
       (throw resp#)
       resp#)))

(defmacro <!!threadx
  "Takes the return value of the thread body as if by <!!, or throws any exception thrown by the body."
  [& body]
  `(let [resp# (<!! (threadx ~@body))]
     (if (instance? Exception resp#)
       (throw resp#)
       resp#)))