core-async-pipeline-bug.clj

core-async-pipeline-bug.clj

(require '[clojure.test :refer :all])
(require '[clojure.core.async :as async])

(deftest from-channel-still-consumed-after-to-channel-is-closed
  (let [from-counter (atom 0)
        xf-counter   (atom 0)
        from         (async/chan)
        to           (async/chan 1000)
        xf           (map #(do (Thread/sleep 100)
                               (swap! xf-counter inc)
                               %))]

    ; start the pipeline
    (async/pipeline 1 to xf from)

    (async/go-loop []
      (async/>! from (swap! from-counter inc))
      (recur))

    ; wait a bit, then close the to channel
    (async/<!! (async/timeout 1000))
    (async/close! to)

    ; comments on pipeline say that the from 
    ; channel should stop being consumed
    ; because the to channel was closed
    (let [from-snapshot (deref from-counter)
          xf-snapshot   (deref xf-counter)]

      (async/<!! (async/timeout 200))
      ; this fails
      (is (= from-snapshot (deref from-counter)))
      (is (= xf-snapshot (deref xf-counter)))

      (async/<!! (async/timeout 400))
      ; this fails (with higher counters)
      (is (= from-snapshot (deref from-counter)))
      (is (= xf-snapshot (deref xf-counter)))

      (async/<!! (async/timeout 600))
      ; this fails (with higher counters)
      (is (= from-snapshot (deref from-counter)))
      (is (= xf-snapshot (deref xf-counter))))))