Fork me on GitHub
Ferdinand Beyer19:02:53

Pub loses items when a channel subscribes to more than one topic. I was chasing a bug in my code today and could shrink it down to the following reproducing code:

(dotimes [n 100]
    (let [start  (async/chan)
          pub    (async/pub start first)
          end    (async/chan)
          _      (async/sub pub :a end)
          _      (async/sub pub :b end)
          recv   (atom #{})
          done   (async/go-loop []
                   (when-let [v (async/<! end)]
                     (swap! recv conj v)
          sent   [[:a 1] [:a 2] [:b 3]]]
      (doseq [v sent]
        (async/>!! start v))
      (async/close! start)
      (async/<!! done)
      (let [exp (set sent)
            act (set @recv)]
        (when (not= exp act)
          (println "No." n)
          (println "Expected:" exp)
          (println "Actual:  " act)
          (assert false)))))
This fails for me at random in one of the first iterations. I think this might be a race condition with the two mults in the pub, one trying to put! an item on a channel that the other one closes. Seems like if I want guaranteed delivery I should not subscribe channels to multiple topics. Is that a bug?