core.clj

(ns bogus-speed-test.core
  (:require [immutant.messaging :as m]))
 
(def thread-done-count (atom 0))
(def q "some.queue")
 
(def listener (atom nil))
 
(m/start q :durable false)
 
(defn unlisten []
  (m/unlisten @listener))
 
(defn listen [consumer-threads]
  (reset! listener
          (m/listen q #(when (= % "done")
                                 (println "done" @thread-done-count)
                                 (swap! thread-done-count inc)) :concurrency consumer-threads)))
 
(defn wait [producer-threads]
  ;;wait for them all to be processed
  (while (> producer-threads @thread-done-count)
    (Thread/sleep 10)))

(defn produce [producer-threads total-messages]
  (dotimes [n producer-threads]
    (future
      (immutant.messaging.core/with-connection {}
        (dotimes [n (/ total-messages producer-threads)]
          (m/publish q "a" :encoding :text :persistent false)))
      (m/publish q "done" :encoding :text :persistent false))))
 
(defn run [consumers producers messages]
  (reset! thread-done-count 0)
  (listen consumers)
  (time
   (do
     (produce producers messages)
     (wait producers)))
  (unlisten))