# CloudAMQP with Clojure

We recommend [Langohr](https://github.com/michaelklishin/langohr), a Clojure client built on the RabbitMQ Java client. Create a [Leiningen](https://leiningen.org) project with this `project.clj` (its `:main` namespace is the one below):

```clojure
(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`:

```clojure
(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 BROKER_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 "BROKER_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 world!")
                    (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](https://github.com/cloudamqp/examples/tree/main/clojure/langohr).

## 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.