I'm learning how to use rabbitmq in the course of writing a webapp in Clojure that will need to call out to Python for occasional machine learning. On the Clojure side, the main library is Langohr, which for some reason doesn't make use of core.async.

For my ultimate use case, I expect the Clojure-side to only ever have to handle one message at a time, immediately on arrival. (The Python side, not shown, will be doing possibly time-consuming machine learning, and will occasionally just send back some weights to the Clojure-side, which will immediately be put to use.) So I don't expect to need any buffering or anything like that, and haven't put them in this proof-of-concept code. I also expect there to be only one producer and only one consumer for each channel.

Toy code connecting the two below, but here are some general questions first:

  1. It seems sensible to me as a design choice to stick incoming and outgoing messages on a core-async channel, just because that's the standard choice, but is it over-engineering to do so when I could just use Langohr functions to listen and pass callbacks?

  2. I'm using go loops and non-blocking puts and takes for everything except for the piece that actually waits to see if an external message is waiting on the channel and sends it out for processing; that one is a blocking take in a thread on the theory that everyone says that if something might sit for a while it should be a thread rather than a go-loop (though I confess to not quite understanding why). Is this wise?

With no further ado, some toy code that demonstrates the design. (Leaving off the namespace declaration, but, obvs, go, chan, go-loop, and thread plus all the angle bracket/bang functions come from core.async. )

 (defn say [outqueue async-channel message]
    (>! async-channel {:queue outqueue :content message})))

(defn dispatch-messages! [mq-channel async-channel]
  (go-loop []
    (let [{:keys [queue content]} (<! async-channel)]
      (lb/publish mq-channel "" queue (json/write-str content) {:content-type "application/json"}))

(defn message-receiver
  [async-channel mq-channel metadata ^bytes payload]
  (let [message (json/read-str (String. payload "UTF-8"))]
      (>! async-channel message))))

(defn listen! [async-channel mq-channel queue]
  (let [handler (partial message-receiver async-channel)]
    (lc/subscribe mq-channel queue handler {:auto-ack true})))

(defn process-messages! [async-channel processor]
    (loop []
      (let [message (<!! async-channel)]
        (processor message))

(defn setup-async-comm [mq-channel, inqueue, outqueue]
  (let [receiver-async-chan (chan)
        sender-async-chan (chan)
        sender-func (partial say outqueue sender-async-chan)
        receiver-func (partial process-messages! receiver-async-chan)]
    (dispatch-messages! mq-channel sender-async-chan)
    (lq/declare mq-channel inqueue {:exclusive false :auto-delete false})
    (lq/declare mq-channel outqueue {:exclusive false :auto-delete false})
    (listen! receiver-async-chan mq-channel inqueue)
    {:send! sender-func :receive! receiver-func}))

(defn answer [msg]
  (println "Clojure takes your Python and doubles it. "
           (mapv #(* 2 %) msg)))

(defn async-comm [mq-channel]
  (let [{:keys [send! receive!]} (setup-async-comm mq-channel "py2clj" "clj2py")]
    (send! [9, 10, 11])
    (receive! answer)))

For purposes of the toy code, the message queue channel (as opposed to the core async channel) is passed in from the main function in another namespace into the call at the bottom, which sets up and fires a test message. Obviously, in production code, the setup will be broken out and the messages will come from user interaction etc. But this demonstrates the basic idea of abstracting away from the messaging details and having, at bottom, a setup function that just returns functions to send and functions to receive, which then can be used by other code at will.

  • \$\begingroup\$ I expect that Langohr does not use core.async to be able to hand through the guarantees of using a Message Queue to the calling code, namely that posting a message is "atomic" and either fails or succeeds. An async channel provides more room for unforseen failure. \$\endgroup\$ – Vogel612 Jul 28 '17 at 12:07
  • \$\begingroup\$ Oh, that's interesting. I didn't know message queues had that kind of guarantee. What does that mean in this context? It's not like a database update where you're sometimes doing two things at once and you need them to succeed or fail together...? \$\endgroup\$ – Paul Gowder Jul 28 '17 at 13:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.