core.clj

(ns async.core
  (:require [clojure.core.async :as async :refer :all]
            [clojure.core.async.lab :as lab]
            [clojure.data.generators :as gen])
  (:gen-class))

;; A Message carries some text to be communicated
;; and a wake channel which notifies the originator
;; that it may transmit again.
(defrecord Message [text wake])

;; Trivial macro that executes its body in an
;; infinite loop.
(defmacro forever [& body]
  `(while true
     ~@body))

(defn- snooze
  "Takes a nap of a (short) random duration"
  []
  (Thread/sleep (int (* 500 (rand)))))

(defn boring
  "Print some pseudo-random noise on a channel"
  [talk]
  (let [wake (chan)
        noise #(str % ": " (gen/string "*" (inc (rand-int 15))))]
    (go
     (doseq [n (iterate inc 1)]
       (>! talk (->Message (noise n) wake))
       (snooze)
       (<! wake)))
    talk))

(defn fan-in
  "Allow whoever is ready to speak up"
  [& in-channels]
  (let [merged (chan)]
    (go
     (forever
      (let [[result _] (alts! in-channels)]
        (>! merged result))))
    merged))

(defn -main
  "Experiment with core/async channels"
  [& args]
  (let [data (fan-in (boring (chan))
                     (boring (chan)))]
    (dotimes [n 5]
      (let [msg1 (<!! data)
            msg2 (<!! data)]
        (println (:text msg1))
        (println (:text msg2))
        (>!! (:wake msg1) true)
        (>!! (:wake msg2) true))))
  (println "\nYou're both boring, I'm leaving."))