demonstrating safety of using close! on sub channels without unsubscribing

experiment.clj

(ns pubsub.experiment
  (:require [clojure.core.async :as >]))

(def publisher (>/chan))

(def p (>/pub publisher (constantly nil)))

(def answers (>/chan))

(defn consume
  [label count]
  (let [s (>/sub p nil (>/chan))]
    (>/go-loop [i 0]
      (if (> i count)
        (>/close! s)
        (do
          (>/>! answers [label (>/<! s)])
          (recur (inc i)))))))

(defn -main
  [& args]
  (println "running pubsub experiment")
  (consume :a 1)
  (consume :b 2)
  (consume :c 3)
  (consume :d 4)
  (consume :e 5)
  (>/go-loop []
    (println "result: " (>/<! answers))
    (recur))
  (>/<!! (>/onto-chan publisher (range 10)))
  (System/exit 0))

output

$ java -cp ~/bin/bench.jar:. clojure.main -m pubsub.experiment
running pubsub experiment
result:  [:a 0]
result:  [:b 0]
result:  [:c 0]
result:  [:d 0]
result:  [:e 0]
result:  [:a 1]
result:  [:b 1]
result:  [:c 1]
result:  [:d 1]
result:  [:e 1]
result:  [:b 2]
result:  [:c 2]
result:  [:d 2]
result:  [:e 2]
result:  [:c 3]
result:  [:d 3]
result:  [:e 3]
result:  [:d 4]
result:  [:e 4]
result:  [:e 5]