Introduction to Clojure's core.async library

core-async-demo.clj

;;;;;;;;;;
;;;; Introduction to core.async based on
;;;; https://github.com/clojure/core.async/tree/master/examples
;;;;;;;;;;

(require [clojure.core.async
          :refer [>! <! >!! <!! go chan close! alts! alts!! alt! alt!!
                  timeout dropping-buffer sliding-buffer]])

;;;;;;;;;;
;;; Channels
;;;;;;;;;;

;; Channels can be unbuffered (which block if there's no "rendezvous",
;; i.e. there must be someone who puts to and someone who takes from the
;; channel).
(chan)

;; Or buffered (which block if the buffer is full).
(chan 13)

;;;;;;;;;;
;;; Threads
;;;;;;;;;;

;; In ordinary threads channels can be created in any matter. close! stops a
;; channel to accept puts (>!!), but remaining values are still available to
;; take (<!!). Drained channels return nil on take.
(let [c (chan 1)]
  (>!! c "foo")
  (println (<!! c))
  (close! c))

;; The go macro turns its body into a state machine which is executed in a
;; special thread pool. Calls to blocking operations such as >! or <! are
;; interpreted as transitions where the calling thread can be 'parked'. In order
;; to do so, go examines its body for lambda expressions which are considered as
;; potentially blocking. Inside go blocks >! and <! are used for puts
;; respectively takes.
(let [c (chan)]
  (go (>! c "foo"))
  (println (<!! c))
  (close! c))

;;;;;;;;;;
;;; alt and alts
;;;;;;;;;;

;; Similar to Go's select core.async provides a non-deterministic selection over
;; an arbitrary number of channels: alts! (or alts!! outside of go
;; blocks). Additionally, it provides options for :default values.
(let [c1 (chan)
      c2 (chan)]
  (go (doseq [_ (range 3)]
        (let [[value _] (alts! [c1 c2] :default "baz")]
          (println value))))
  (>!! c1 "foo")
  (>!! c2 "bar"))

;; Whereas alts! and alts!! are functions, alt! and alt!! are macros that make a
;; single choice between one of several channels that must be known statically,
;; but can be choosen with a :priority (if true, ordering matters). Furthermore,
;; timeout creates a channel that waits for a specific amount of ms and then
;; closes.
(let [blocking-channel (chan)
      timeout-channel (timeout 1000)]
  (alt!!
    blocking-channel "You ain't gonna see this."
    timeout-channel "But this."))

;;;;;;;;;;
;;; Special channels
;;;;;;;;;;

;; Drops newest values when full.
(chan (dropping-buffer 10))

;; Drops oldest values when full.
(chan (sliding-buffer 10))

;;;;;;;;;;
;;; Example 1.
;;;;;;;;;;

(let [my-data [1 3 3 7]
      my-channel (chan)
      my-barrier (chan)]
  (println "Main thread started.")

  (go
    (doseq [val my-data]
      (>! my-channel val)
      (Thread/sleep 1000))
    (>! my-barrier 1))

  (go
    (while true
      (println (<! my-channel))))
    
  (<!! my-barrier)
  (println "Main thread finished."))

;;;;;;;;;;
;;; Example 2.
;;;;;;;;;;

(let [c1 (chan)
      c2 (chan)
      send-async (fn [delay channel id]
                   (go
                     (Thread/sleep delay)
                     (>! channel id)))]
    
  (send-async 3000 c1 1)
  (send-async 2000 c2 2)

  (let [[value _] (alts!! [c1 c2 (timeout 1000)])]
    (if value 
      (println (str "Received id " value "."))
      (println "Timed out."))))

;;;;;;;;;;
;;; Example 3.
;;;;;;;;;;

(time 
  (let [n 1000
        channels (repeatedly n chan)] 

  (doseq [c channels] (go (>! c "hi")))

  (dotimes [_ n]
    (let [[value _] (alts!! channels)]
      (assert (= "hi" value))))))