lib.rabbitmq

RabbitMQ management, routing, publishing, and consumers

Manage queues, exchanges, bindings, vhosts, network state, publishing, and consumers through a component-friendly client.

All libraries

1    Overview

The main client combines connection settings for RabbitMQ and its management API. High-level functions return compact maps for queues, exchanges, bindings, consumers, routing, and network state.

2    Walkthrough

2.1    Create a client

(require '[lib.rabbitmq :as rabbitmq])

(def mq
  (rabbitmq/rabbit
   {:host "localhost"
    :port 5672
    :management-port 15672
    :vhost "/"}))

2.2    Create routing

(-> mq
    (rabbitmq/add-exchange
     "events"
     {:type "topic" :durable true})
    (rabbitmq/add-queue
     "events.audit"
     {:durable true})
    (rabbitmq/bind-queue
     "events"
     "events.audit"
     {:routing-key "audit.*"}))

(rabbitmq/routing mq)

2.3    Publish and consume

(rabbitmq/publish mq
                  "events"
                  "created"
                  {:key "audit.created"})

(rabbitmq/connect mq)
(rabbitmq/consume mq "events.audit" handle-message)

3    API

*default-exchanges* ^

NONE
(def ^:dynamic *default-exchanges*
  #{"" "amq.direct" "amq.fanout" "amq.headers"
    "amq.match" "amq.rabbitmq.trace" "amq.rabbitmq.log" "amq.topic"})
link

*default-options* ^

NONE
(def ^:dynamic *default-options*)
link

adapt ^

[{:keys [function]} channel]
Added 4.1.4

creates an adaptor for consuming input

v 4.1.4
(defn adapt
  [{:keys [function]} channel]
  (proxy [DefaultConsumer] [channel]
    (handleDelivery [tag envelope properties body]
      (function (String. body)))
    (handleCancel [tag])
    (handleCancelOk [tag])))
link
(let [out (atom nil)
      consumer (adapt {:function #(reset! out %)} nil)]
  [(.handleDelivery ^DefaultConsumer consumer "tag" nil nil (.getBytes "hello"))
   @out
   (instance? Consumer consumer)])
=> [nil "hello" true]

add-exchange ^

[mq name] [mq name opts]
Added 4.1.4

adds an exchange and returns the client

v 4.1.4
(defn add-exchange
  ([mq name]
   (add-exchange mq name {}))
  ([mq name opts]
   (api/add-exchange mq name opts)
   mq))
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/add-exchange (fn [_ name opts] (swap! calls conj [name opts]))]
    [(add-exchange sample-rabbit "ex1" {:type "topic"})
     @calls]))
=> [sample-rabbit
    [["ex1" {:type "topic"}]]]

add-queue ^

[mq name] [mq name opts]
Added 4.1.4

adds a queue and returns the client

v 4.1.4
(defn add-queue
  ([mq name]
   (add-queue mq name {}))
  ([mq name opts]
   (api/add-queue mq name opts)
   mq))
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/add-queue (fn [_ name opts] (swap! calls conj [name opts]))]
    [(add-queue sample-rabbit "q1" {:durable true})
     @calls]))
=> [sample-rabbit
    [["q1" {:durable true}]]]

bind-exchange ^

[mq source dest] [mq source dest opts]
Added 4.1.4

binds an exchange and returns the client

v 4.1.4
(defn bind-exchange
  ([mq source dest]
   (bind-exchange mq source dest {}))
  ([mq source dest opts]
   (api/bind-exchange mq source dest opts)
   mq))
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/bind-exchange (fn [_ source dest opts]
                                                 (swap! calls conj [source dest opts]))]
    [(bind-exchange sample-rabbit "ex1" "ex2" {:routing-key "a"})
     @calls]))
=> [sample-rabbit
    [["ex1" "ex2" {:routing-key "a"}]]]

bind-queue ^

[mq source dest] [mq source dest opts]
Added 4.1.4

binds a queue and returns the client

v 4.1.4
(defn bind-queue
  ([mq source dest]
   (bind-queue mq source dest {}))
  ([mq source dest opts]
   (api/bind-queue mq source dest opts)
   mq))
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/bind-queue (fn [_ source dest opts]
                                              (swap! calls conj [source dest opts]))]
    [(bind-queue sample-rabbit "ex1" "q1" {:routing-key ""})
     @calls]))
=> [sample-rabbit
    [["ex1" "q1" {:routing-key ""}]]]

connect ^

[{:keys [host port username password vhost heartbeat timeout recovery-interval topology-recovery]}]
Added 4.1.4

connects the consumer to the rabbitmq instance

v 4.1.4
(defn connect
  [{:keys [host port username password vhost heartbeat timeout
           recovery-interval topology-recovery]}]
  (let [cfactory (ConnectionFactory.)]
    (doto cfactory
      (.setUsername ^String username)
      (.setPassword ^String password)
      (.setHost ^String host)
      (.setPort ^int port)
      (.setVirtualHost ^String vhost)
      (.setRequestedHeartbeat ^int heartbeat)
      (.setConnectionTimeout ^int timeout)
      (.setNetworkRecoveryInterval ^int recovery-interval)
      (.setTopologyRecoveryEnabled ^boolean topology-recovery))
    (.newConnection cfactory (str "lib.rabbitmq:" port))))
link
(connect (assoc *default-options*
                :host "127.0.0.1"
                :port 1
                :timeout 10))
=> (throws Exception)

consume ^

[channel queue autoack tag {:keys [id], :as handler}]
Added 4.1.4

creates a consume function

v 4.1.4
(defn consume
  [channel queue autoack tag {:keys [id] :as handler}]
  (.basicConsume channel
                 ^String (name queue)
                 ^Boolean autoack
                 ^String (name (or id tag))
                 ^Consumer (adapt handler channel)))
link
(let [{:keys [calls channel]} (proxy-channel)]
  [(consume channel :queue-a true :tag-a {:id :consumer-a
                                          :function identity})
   @calls])
=> #(let [[tag calls] %
          [[name args]] calls]
      (and (= "consumer-tag" tag)
           (= "basicConsume" name)
           (= "queue-a" (nth args 0))
           (= true (nth args 1))
           (= "consumer-a" (nth args 2))
           (instance? Consumer (nth args 3))))

create-client ^

[] [m]
Added 4.1.4

creates a rabbitmq client map

v 4.1.4
(defn create-client
  ([] (create-client {}))
  ([m]
   (let [m (merge *default-options* m)]
     (assoc m :vhost-encode (vhost-encode (:vhost m))))))
link
(select-keys sample-rabbit [:protocol :host :management-port :vhost-encode])
=> {:protocol "http"
    :host "localhost"
    :management-port 15672
    :vhost-encode "%2F"}

delete-exchange ^

[mq name]
Added 4.1.4

deletes an exchange and returns the client

v 4.1.4
(defn delete-exchange
  [mq name]
  (api/delete-exchange mq name)
  mq)
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/delete-exchange (fn [_ name] (swap! calls conj name))]
    [(delete-exchange sample-rabbit "ex1")
     @calls]))
=> [sample-rabbit ["ex1"]]

delete-queue ^

[mq name]
Added 4.1.4

deletes a queue and returns the client

v 4.1.4
(defn delete-queue
  [mq name]
  (api/delete-queue mq name)
  mq)
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/delete-queue (fn [_ name] (swap! calls conj name))]
    [(delete-queue sample-rabbit "q1")
     @calls]))
=> [sample-rabbit ["q1"]]

install-vhost ^

[{:keys [vhost username], :as rabbitmq}]
Added 4.1.4

installs a missing vhost and permissions

v 4.1.4
(defn install-vhost
  [{:keys [vhost username] :as rabbitmq}]
  (let [curr (->> (api/list-vhosts rabbitmq)
                  (map :name)
                  set)]
    (when-not (curr vhost)
      (api/add-vhost rabbitmq vhost {})
      (api/add-permissions rabbitmq username {:configure ".*"
                                              :write ".*"
                                              :read ".*"})))
  rabbitmq)
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/list-vhosts (fn [_] [{:name "existing"}])
                lib.rabbitmq.api/add-vhost (fn [_ vhost body] (swap! calls conj [:vhost vhost body]))
                lib.rabbitmq.api/add-permissions (fn [_ user body] (swap! calls conj [:permissions user body]))]
    [(install-vhost (assoc sample-rabbit :vhost "new" :username "guest"))
     @calls]))
=> [(assoc sample-rabbit :vhost "new" :username "guest")
    [[:vhost "new" {}]
     [:permissions "guest" {:configure ".*" :write ".*" :read ".*"}]]]

list-bindings ^

[mq]
Added 4.1.4

groups bindings by source and destination type

v 4.1.4
(defn list-bindings
  [mq]
  (->> (api/list-bindings mq)
       (remove #(-> % :source empty?))
       (reduce (fn [out {:keys [source destination destination-type] :as data}]
                 (update-in out [source (keyword (str destination-type "s")) destination]
                            (fnil #(conj % (dissoc data :source :vhost :destination :destination-type))
                                  [])))
               {})))
link
(with-redefs [lib.rabbitmq.api/list-bindings (fn [_]
                                               [{:source "ex1" :destination "q1" :destination-type "queue" :routing-key ""}
                                                {:source "ex1" :destination "ex2" :destination-type "exchange" :routing-key "a"}])]
  (list-bindings sample-rabbit))
=> {"ex1" {:queues {"q1" [{:routing-key ""}]}
           :exchanges {"ex2" [{:routing-key "a"}]}}}

list-consumers ^

[mq]
Added 4.1.4

lists consumers grouped by queue

v 4.1.4
(defn list-consumers
  [mq]
  (->> (api/list-consumers mq)
       (map (fn [m]
              {:queue (-> m :queue :name)
               :id (:consumer-tag m)
               :details (:channel-details m)}))
       (reduce (fn [out {:keys [queue id details]}]
                 (assoc-in out [queue (keyword id)] details))
               {})))
link
(with-redefs [lib.rabbitmq.api/list-consumers (fn [_]
                                                [{:queue {:name "q1"}
                                                  :consumer-tag "c1"
                                                  :channel-details {:peer-host "host"}}])]
  (list-consumers sample-rabbit))
=> {"q1" {:c1 {:peer-host "host"}}}

list-exchanges ^

[mq]
Added 4.1.4

filters default exchanges from exchange listings

v 4.1.4
(defn list-exchanges
  [mq]
  (->> (api/list-exchanges mq)
       (remove #(-> % :name *default-exchanges*))
       (reduce (fn [out {:keys [name] :as data}]
                 (assoc out name (select-keys data [:type :internal :auto-delete :durable])))
               {})))
link
(with-redefs [lib.rabbitmq.api/list-exchanges (fn [_]
                                                [{:name "amq.direct" :type "direct"}
                                                 {:name "ex1" :type "topic" :durable true}])]
  (list-exchanges sample-rabbit))
=> {"ex1" {:type "topic" :durable true}}

list-queues ^

[mq]
Added 4.1.4

lists queues as a keyed map

v 4.1.4
(defn list-queues
  [mq]
  (->> (api/list-queues mq)
       (reduce (fn [out {:keys [name] :as data}]
                 (assoc out name (select-keys data [:exclusive :auto-delete :durable])))
               {})))
link
(with-redefs [lib.rabbitmq.api/list-queues (fn [_]
                                             [{:name "q1" :durable true}
                                              {:name "q2" :exclusive true}])]
  (list-queues sample-rabbit))
=> {"q1" {:durable true}
    "q2" {:exclusive true}}

network ^

[rabbitmq]
Added 4.1.4

returns the rabbitmq network summary

v 4.1.4
(defn network
  [rabbitmq]
  (let [vhosts (->> (api/list-vhosts rabbitmq)
                    (mapv :name))
        connections (->> (api/list-connections rabbitmq)
                         (map #(select-keys % [:name :peer-port :peer-host :host :port])))
        channels (->> (api/list-channels rabbitmq)
                      (reduce (fn [out data]
                                (update-in out [(-> data :connection-details :name)] (fnil conj #{}) data))
                              {}))
        cluster-name (:name (api/cluster-name rabbitmq))
        nodes (->> (api/list-nodes rabbitmq)
                   (mapv :name))]
    {:cluster-name cluster-name
     :nodes nodes
     :vhosts vhosts
     :connections connections
     :channels channels}))
link
(with-redefs [lib.rabbitmq.api/list-vhosts (fn [_] [{:name "/"}])
              lib.rabbitmq.api/list-connections (fn [_] [{:name "conn" :peer-port 1 :peer-host "a" :host "b" :port 2}])
              lib.rabbitmq.api/list-channels (fn [_] [{:connection-details {:name "conn"} :name "ch1"}])
              lib.rabbitmq.api/cluster-name (fn [_] {:name "rabbit@node"})
              lib.rabbitmq.api/list-nodes (fn [_] [{:name "rabbit@node"}])]
  (network sample-rabbit))
=> {:cluster-name "rabbit@node"
    :nodes ["rabbit@node"]
    :vhosts ["/"]
    :connections [{:name "conn" :peer-port 1 :peer-host "a" :host "b" :port 2}]
    :channels {"conn" #{{:connection-details {:name "conn"} :name "ch1"}}}}

publish ^

[mq exchange message] [mq exchange message opts]
Added 4.1.4

publishes a message through the management api

v 4.1.4
(defn publish
  ([mq exchange message]
   (publish mq exchange message {}))
  ([mq exchange message opts]
   (api/add-message mq exchange {:routing-key (or (:key opts) "")
                                 :payload message
                                 :payload-encoding (or (:encoding opts) "string")
                                 :properties (dissoc opts :key :encoding)})))
link
(let [calls (atom [])]
  (with-redefs [lib.rabbitmq.api/add-message (fn [_ exchange body]
                                               (swap! calls conj [exchange body])
                                               {:status "ok"})]
    [(publish sample-rabbit "ex1" "hello" {:key "a"})
     @calls]))
=> [{:status "ok"}
    [["ex1" {:routing-key "a"
             :payload "hello"
             :payload-encoding "string"
             :properties {}}]]]

rabbit ^

[] [m]
Added 4.1.4

creates a rabbitmq record

v 4.1.4
(defn rabbit
  ([] (rabbit {}))
  ([m]
   (-> (map->RabbitMQ (create-client m))
       (component/start))))
link
(-> (rabbit {:host "rabbit"})
    (select-keys [:host :management-port :vhost-encode]))
=> {:host "rabbit"
    :management-port 15672
    :vhost-encode "%2F"}

routing ^

[mq]
Added 4.1.4

returns the routing summary for a vhost

v 4.1.4
(defn routing
  [mq]
  {:queues (list-queues mq)
   :exchanges (list-exchanges mq)
   :bindings (list-bindings mq)})
link
(with-redefs [list-queues (fn [_] {"q1" {:durable true}})
              list-exchanges (fn [_] {"ex1" {:type "topic"}})
              list-bindings (fn [_] {"ex1" {:queues {"q1" [{}]}}})]
  (routing sample-rabbit))
=> {:queues {"q1" {:durable true}}
    :exchanges {"ex1" {:type "topic"}}
    :bindings {"ex1" {:queues {"q1" [{}]}}}}

routing-all ^

[rabbitmq opts]
Added 4.1.4

returns routing summaries for all vhosts

v 4.1.4
(defn routing-all
  [rabbitmq opts]
  (let [vhosts (->> (api/list-vhosts rabbitmq)
                    (mapv :name))]
    (->> vhosts
         (map (fn [vhost]
                [vhost (routing (assoc rabbitmq
                                       :vhost vhost
                                       :vhost-encode (vhost-encode vhost)))]))
         (into {}))))
link
(with-redefs [lib.rabbitmq.api/list-vhosts (fn [_] [{:name "/"} {:name "app"}])
              routing (fn [mq] {:queues {(str (:vhost mq)) {}}})]
  (routing-all sample-rabbit {}))
=> {"/" {:queues {"/" {}}}
    "app" {:queues {"app" {}}}}

vhost-encode ^

[vhost]
Added 4.1.4

encodes vhosts for the management api

v 4.1.4
(defn vhost-encode
  [vhost]
  (URLEncoder/encode (str vhost) "UTF-8"))
link
(vhost-encode "/")
=> "%2F"

*default-methods* ^

NONE
(def ^:dynamic *default-methods*)
link

*methods* ^

NONE
(def ^:dynamic *methods*)
link

+init+ ^

NONE
(defonce +init+ (create-api-functions *methods*))
link

build-args ^

[args]
Added 4.1.4

builds link argument vectors

v 4.1.4
(defn build-args
  [args]
  (mapv (fn [[t data]]
          (if (= t :entry)
            (symbol (second data))
            data))
        args))
link
(build-args [[:string "hello"]
             [:entry [1 "world"]]
             [:keyword '(:hello rabbitmq)]])
=> ["hello" 'world '(:hello rabbitmq)]

classify-args ^

[s]
Added 4.1.4

classifies api link fragments

v 4.1.4
(defn classify-args
  [s]
  (cond (and (.startsWith s "{")
             (.endsWith s "}"))
        (let [s (subs s 1 (dec (count s)))]
          (cond (.startsWith s "%")
                (let [s (subs s 1)
                      [num name] (str/split s #":")]
                  [:entry [(Integer/parseInt num) name]])

                :else
                [:keyword (list (keyword s) 'rabbitmq)]))

        :else
        [:string s]))
link
[(classify-args "hello")
 (classify-args "{hello}")
 (classify-args "{%1:hello}")]
=> [[:string "hello"]
    [:keyword '(:hello rabbitmq)]
    [:entry [1 "hello"]]]

create-accessor-form ^

[fname {:keys [link methods]}]
Added 4.1.4

creates accessor forms

v 4.1.4
(defn create-accessor-form
  [fname {:keys [link methods]}]
  (let [args (link-args link)
        getter-key (or (:getter methods) :get)
        getter-form (create-link-form args getter-key)
        setter-form (when-let [setter-key (:setter methods)]
                      (create-body-form args setter-key))]
    `(defn ~(symbol (if (keyword? fname) (name fname) fname))
       ~@(filter identity [getter-form setter-form]))))
link
(create-accessor-form :cluster-name
                      {:link "cluster-name"
                       :methods {:setter :put}})
=> '(clojure.core/defn cluster-name
      ([rabbitmq]
       (lib.rabbitmq.request/request
        rabbitmq (std.string/joinl ["cluster-name"] "/")
        :get))
      ([rabbitmq body]
       (lib.rabbitmq.request/request
        rabbitmq (std.string/joinl ["cluster-name"] "/")
        :put {:body body})))

create-api-functions ^

[methods]
Added 4.1.4

creates the api function set

v 4.1.4
(defn create-api-functions
  [methods]
  (mapv (fn [[name opts]]
          (eval (if (= :form (:type opts))
                  `(do ~@(create-function-forms name opts))
                  (create-accessor-form name opts))))
        (seq methods)))
link
(-> (create-api-functions {:tmp-route {:link "overview"}})
    count)
=> 1

create-body-form ^

[{:keys [inputs vargs]} key]
Added 4.1.4

creates a body request form

v 4.1.4
(defn create-body-form
  [{:keys [inputs vargs]} key]
  (list (conj (vec (cons 'rabbitmq vargs)) 'body)
        (list `request/request 'rabbitmq (list `str/joinl inputs "/") key {:body 'body})))
link
(create-body-form '{:inputs ["hello" (:world rabbitmq) foo bar]
                    :vargs [foo bar]}
                  :post)
=> '([rabbitmq foo bar body]
     (lib.rabbitmq.request/request
      rabbitmq
      (std.string/joinl ["hello" (:world rabbitmq) foo bar] "/")
      :post
      {:body body}))

create-function-forms ^

[fname {:keys [link methods]}]
Added 4.1.4

creates CRUD forms for routes

v 4.1.4
(defn create-function-forms
  [fname {:keys [link methods]}]
  (let [args (link-args link)
        fname (if (keyword? fname) (name fname) fname)
        forms {:get (create-link-form args :get)
               :delete (create-link-form args :delete)
               :put (create-body-form args :put)
               :post (create-body-form args :post)}
        prefix {:get :get :delete :delete :put :add :post :add}]
    (reduce (fn [arr k]
              (conj arr `(defn ~(symbol (str (name (get prefix k)) "-" fname))
                           ~(get forms k))))
            []
            methods)))
link
(->> (create-function-forms :vhost
                            {:type :form
                             :link "vhosts/{%1:vhost}"
                             :methods #{:get :put :delete}})
     (map second))
=> '(get-vhost delete-vhost add-vhost)

create-link-form ^

[{:keys [inputs vargs]} key]
Added 4.1.4

creates a link request form

v 4.1.4
(defn create-link-form
  [{:keys [inputs vargs]} key]
  (list (vec (cons 'rabbitmq vargs))
        (list `request/request 'rabbitmq (list `str/joinl inputs "/") key)))
link
(create-link-form '{:inputs ["hello" (:world rabbitmq) foo bar]
                    :vargs [foo bar]}
                  :delete)
=> '([rabbitmq foo bar]
     (lib.rabbitmq.request/request
      rabbitmq
      (std.string/joinl ["hello" (:world rabbitmq) foo bar] "/")
      :delete))

link-args ^

[uri]
Added 4.1.4

parses link templates

v 4.1.4
(defn link-args
  [uri]
  (let [args (->> (str/split uri #"/")
                  (map classify-args))
        entries (->> args
                     (filter #(-> % first (= :entry)))
                     (sort-by #(-> % second first)))]
    {:inputs (build-args args)
     :vargs (mapv (comp symbol second second) entries)}))
link
(link-args "hello/{world}/{%1:foo}/{%2:bar}")
=> {:inputs ["hello" '(:world rabbitmq) 'foo 'bar]
    :vargs ['foo 'bar]}

spec ^

NONE
(def spec)
link

*default-options* ^

NONE
(def ^:dynamic *default-options*)
link

adapt ^

[{:keys [function]} channel]
Added 4.1.4

creates an adaptor for consuming input

v 4.1.4
(defn adapt
  [{:keys [function]} channel]
  (proxy [DefaultConsumer] [channel]
    (handleDelivery [tag envelope properties body]
      (function (String. body)))
    (handleCancel [tag])
    (handleCancelOk [tag])))
link
(let [out (atom nil)
      consumer (adapt {:function #(reset! out %)} nil)]
  [(.handleDelivery ^DefaultConsumer consumer "tag" nil nil (.getBytes "hello"))
   @out
   (instance? Consumer consumer)])
=> [nil "hello" true]

connect ^

[{:keys [host port username password vhost heartbeat timeout recovery-interval topology-recovery]}]
Added 4.1.4

connects the consumer to the rabbitmq instance

v 4.1.4
(defn connect
  [{:keys [host port username password vhost heartbeat timeout
           recovery-interval topology-recovery]}]
  (let [cfactory (ConnectionFactory.)]
    (doto cfactory
      (.setUsername ^String username)
      (.setPassword ^String password)
      (.setHost ^String host)
      (.setPort ^int port)
      (.setVirtualHost ^String vhost)
      (.setRequestedHeartbeat ^int heartbeat)
      (.setConnectionTimeout ^int timeout)
      (.setNetworkRecoveryInterval ^int recovery-interval)
      (.setTopologyRecoveryEnabled ^boolean topology-recovery))
    (.newConnection cfactory (str "lib.rabbitmq:" port))))
link
(connect (assoc *default-options*
                :host "127.0.0.1"
                :port 1
                :timeout 10))
=> (throws Exception)

consume ^

[channel queue autoack tag {:keys [id], :as handler}]
Added 4.1.4

creates a consume function

v 4.1.4
(defn consume
  [channel queue autoack tag {:keys [id] :as handler}]
  (.basicConsume channel
                 ^String (name queue)
                 ^Boolean autoack
                 ^String (name (or id tag))
                 ^Consumer (adapt handler channel)))
link
(let [{:keys [calls channel]} (proxy-channel)]
  [(consume channel :queue-a true :tag-a {:id :consumer-a
                                          :function identity})
   @calls])
=> #(let [[tag calls] %
          [[name args]] calls]
      (and (= "consumer-tag" tag)
           (= "basicConsume" name)
           (= "queue-a" (nth args 0))
           (= true (nth args 1))
           (= "consumer-a" (nth args 2))
           (instance? Consumer (nth args 3))))

*default-request-options* ^

NONE
(def ^:dynamic *default-request-options*)
link

basic-auth-header ^

[{:keys [username password]}]
Added 4.1.4

creates the basic auth header

v 4.1.4
(defn basic-auth-header
  [{:keys [username password]}]
  (str "Basic "
       (encode/to-base64 (.getBytes (str username ":" password) "UTF-8"))))
link
(basic-auth-header sample-rabbit)
=> "Basic Z3Vlc3Q6Z3Vlc3Q="

create-url ^

[{:keys [protocol host management-port]} suburl]
Added 4.1.4

creates the management url

v 4.1.4
(defn create-url
  [{:keys [protocol host management-port]} suburl]
  (str protocol "://" host ":" management-port "/api/" suburl))
link
(create-url sample-rabbit "hello")
=> "http://localhost:15672/api/hello"

request ^

[rabbit suburl] [rabbit suburl method] [rabbit suburl method opts]
Added 4.1.4

creates a management api request

v 4.1.4
(defn request
  ([rabbit suburl]
   (request rabbit suburl :get))
  ([rabbit suburl method]
   (request rabbit suburl method {}))
  ([rabbit suburl method opts]
   (let [req (collection/merge-nested
              *default-request-options*
              opts
              {:method method
               :headers {"Authorization" (basic-auth-header rabbit)}})
         req (assoc req :uri (create-url rabbit suburl))]
     ((-> http/request
          wrap-generate-json
          wrap-parse-json)
      req))))
link
(let [captured (atom nil)]
  (with-redefs [net.http/request (fn [req]
                                   (reset! captured req)
                                   {:status 200
                                    :body (json/write {:name "rabbit@node"})})]
    [(request sample-rabbit "cluster-name")
     @captured]))
=> [{:name "rabbit@node"}
    {:uri "http://localhost:15672/api/cluster-name"
     :method :get
     :headers {"Accept" "application/json"
               "Content-Type" "application/json"
               "Authorization" "Basic Z3Vlc3Q6Z3Vlc3Q="}}]

update-nested-keys ^

[m func]
Added 4.1.4

updates nested keys recursively

v 4.1.4
(defn update-nested-keys
  [m func]
  (cond (map? m)
        (reduce-kv (fn [out k v]
                     (assoc out
                            (func k)
                            (if (coll? v)
                              (update-nested-keys v func)
                              v)))
                   {}
                   m)

        (coll? m)
        (into (empty m) (map #(update-nested-keys % func) m))

        :else
        m))
link
(update-nested-keys {:a {:b {:c 1}}}
                    #(keyword (str (name %) "-boo")))
=> {:a-boo {:b-boo {:c-boo 1}}}

wrap-generate-json ^

[f]
Added 4.1.4

encodes nested request bodies as snake_case json

v 4.1.4
(defn wrap-generate-json
  [f]
  (fn [request]
    (let [body (:body request)
          body (cond (nil? body) nil
                     (string? body) body
                     :else (-> body
                               (update-nested-keys (comp case/snake-case name))
                               (json/write)))
          request (cond-> request
                    body (assoc :body body))]
      (f request))))
link
((wrap-generate-json identity)
 {:body {:helloWorld {:queueName "q1"}}})
=> {:body "{"hello_world":{"queue_name":"q1"}}"}

wrap-parse-json ^

[f]
Added 4.1.4

parses successful json responses

v 4.1.4
(defn wrap-parse-json
  [f]
  (fn [request]
    (let [res (f request)
          status (:status res)
          body (:body res)]
      (cond (<= 200 status 299)
            (if (or (nil? body) (= "" body))
              (if (= status 204) true res)
              (json/read body json/+keyword-mapper+))

            (<= 400 status)
            (throw (ex-info "HTTP Error" res))

            :else
            res))))
link
((wrap-parse-json identity)
 {:status 200
  :body (json/write {:a 1 :b 2})})
=> {:a 1 :b 2}

Related

jvm.artifactdocs

artifact coordinates and paths

jvm.classloaderdocs

classpath and classloaders

jvm.depsdocs

JVM dependency resolution

jvm.monitordocs

JVM metrics

jvm.namespacedocs

namespace management

jvm.protocoldocs

JVM extension protocols

jvm.reflectdocs

Java reflection

jvm.requiredocs

dependency-aware requiring