ebaxt

(let [[message-seq put] (pipe)]
  (letfn [(message-handler [ch msg-meta payload]
            (put {:ch ch :msg-meta msg-meta :payload payload}))]
    (lc/subscribe ch qname message-handler :auto-ack true))
  message-seq)

ebaxt

(defn pipe
  "Returns a vector containing a sequence that will read from the
   queue, and a function that inserts items into the queue.

   Source: http://clj-me.cgrand.net/2010/04/02/pipe-dreams-are-not-necessarily-made-of-promises/"
  []
  (let [q (LinkedBlockingQueue.)
        EOQ (Object.)
        NIL (Object.)
        s (fn queue-seq []