mapcat testing

core.clj

(ns lab.core
  (:gen-class)
  (:require [clojure.core.async :as a]))

(defn event-proc
  "linearize event logging"
  [chan]
  (a/thread
    (loop []
      (let [{:keys [label data]} (a/<!! chan)
            log (str label " occurred at " (System/currentTimeMillis) ".")
            log (if data (str log " Additional data: " data) log)]
        (when label
          (when (#{:provided :complete} label)
            (println log))
          (recur))))))

(defn event-fn
  [chan]
  (fn
    ([label]
     (a/>!! chan {:label label}))
    ([label data]
     (a/>!! chan {:label label
                  :data  data}))))

(defn consumer
  [chan event]
  (a/thread
    (loop []
      (when-let [v (a/<!! chan)]
        (a/<!! (a/timeout 10))
        (event :consumed v)
        (recur)))))

(defn provider
  [chan lim event]
  (a/<!! (a/onto-chan chan (range lim)))
  (event :provided))

(defn -main
  []
  (let [e          (a/chan)
        event      (event-fn e)
        event-proc (event-proc e)
        c          (a/chan 1 (mapcat range))
        consume    (consumer c event)]
    (provider c 2048 event)
    (a/<!! consume)
    (event :complete)))