CloudAMQP with Clojure

We recommend Langohr, a Clojure client built on the RabbitMQ Java client. Create a Leiningen project with this project.clj (its :main namespace is the one below):

(defproject clojure-langohr-example "1.0.0-SNAPSHOT"
  :description "Example of publishing and consuming messages with Langohr"
  :main clojure-langohr-example.core
  :dependencies [[org.clojure/clojure "1.12.6"]
                 [com.novemberain/langohr "5.6.0"]
                 ;; amqp-client depends on slf4j-api 1.7, which slf4j-simple 2 can't bind to
                 [org.slf4j/slf4j-api "2.0.20"]
                 [org.slf4j/slf4j-simple "2.0.20"]]
  :profiles {:uberjar {:aot :all}})

Publish and consume

Set CLOUDAMQP_URL to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the example connects to a local broker.

Save the connection, publisher and consumer as src/clojure_langohr_example/core.clj:

(ns clojure-langohr-example.core
  (:require [clojure.string :as str]
            [langohr.basic :as lb]
            [langohr.channel :as lch]
            [langohr.confirm :as lcf]
            [langohr.consumers :as lc]
            [langohr.core :as rmq]
            [langohr.queue :as lq])
  (:import [javax.net.ssl SSLContext])
  (:gen-class))

(defn connection-settings
  "Settings for CLOUDAMQP_URL. For amqps:// the JVM's default SSLContext
  verifies the server certificate and hostname; without it Langohr's
  client would trust any certificate."
  [uri]
  (cond-> {:uri uri}
    (str/starts-with? uri "amqps://") (assoc :ssl-context (SSLContext/getDefault)
                                             :verify-hostname true)))

(defn connect []
  (rmq/connect (connection-settings (or (System/getenv "CLOUDAMQP_URL")
                                        "amqp://guest:guest@localhost"))))

(defn publish
  "Publishes a persistent message and waits for the broker's confirm"
  [ch queue message]
  (lq/declare ch queue {:durable true :exclusive false :auto-delete false})
  (lcf/select ch)
  (lb/publish ch "" queue message {:content-type "text/plain" :persistent true})
  (lcf/wait-for-confirms-or-die ch 5000)
  (println (format " [x] Sent '%s'" message)))

(defn consume
  "Consumes with a prefetch of 10, acking each message after handler returns"
  [ch queue handler]
  (lq/declare ch queue {:durable true :exclusive false :auto-delete false})
  (lb/qos ch 10)
  (lc/subscribe ch queue
                (fn [ch {:keys [delivery-tag]} ^bytes payload]
                  (handler (String. payload "UTF-8"))
                  (lb/ack ch delivery-tag))
                {:auto-ack false}))

(defn -main [& [mode]]
  (let [conn (connect)]
    (case mode
      "publish" (do (publish (lch/open conn) "hello" "Hello CloudAMQP!")
                    (rmq/close conn))
      "worker"  (do (consume (lch/open conn) "hello" #(println (format " [x] Received '%s'" %)))
                    (println " [*] Waiting for messages. To exit press CTRL+C"))
      (do (println "Usage: lein run publish|worker")
          (rmq/close conn)
          (System/exit 1)))))

Run lein run worker in one terminal and lein run publish in another.

A runnable example is in clojure/langohr in cloudamqp/examples.

Good to know

Connect with the URL. (rmq/connect {:uri url}), where the URL is the one on your instance’s details page in the CloudAMQP console.
Verify the server certificate. For amqps:// URLs, pass the JVM’s default SSLContext as :ssl-context and set :verify-hostname true, as connection-settings does. Without them, Langohr 5.6 connects with a trust manager that accepts any certificate.
Automatic recovery is on by default: after a network failure the client reconnects and restores channels, queues and consumers.
Use publisher confirms (langohr.confirm/select, wait-for-confirms-or-die) and persistent messages, so a message isn’t lost if the connection or broker goes down.
Ack after processing, and set a prefetch limit with langohr.basic/qos, so a consumer doesn’t buffer more messages than it can handle.

← Java

Go →