Duct RabbitMQ

Integrant methods for connecting to a RabbitMQ server via Langohr.

Installation

To install, add the following to your project :dependencies

[viebel/rabbitmq "0.1.2"]

Usage

This library provides:

a multimethod for :duct.amqp.rabbitmq/langohr that initiates the connection based on the settings.
a multimethod for :duct.amqp.rabbitmq/langohr-consumers that subscribes handlers on queues.
a multimethod for :duct.amqp.rabbitmq/langohr-producers that creates publish functions for producers.

Connection

URI

:duct.amqp.rabbitmq/langohr {:host "localhost"
                             :port 5672
                             :username "guest"
                             :password "guest"
                             :vhost "/"}

Consumers

:duct.amqp.rabbitmq/langohr-consumers {:consumers [{:queue "my-queue"
                                                    :declare-queue {:durable false
                                                                    :exclusive false
                                                                    :auto-delete true
                                                                    :arguments {}}
                                                     :handler #ig/ref :my-prj.handler/handle-msg}]
                                       :connection #ig/ref :duct.amqp.rabbitmq/langohr}
:my-prj.handler/handle-msg {}

Note: You can optionally pass a duct logger. If you do so, a message will be logged for each queue subscriber, when it subscribes.

:duct.amqp.rabbitmq/langohr-consumers {:consumers [{:queue "my-queue"
                                                    :declare-queue {:durable false
                                                                    :exclusive false
                                                                    :auto-delete true
                                                                    :arguments {}}
                                                     :handler #ig/ref :my-prj.handler/handle-msg}]
                                       :connection #ig/ref :duct.amqp.rabbitmq/langohr
                                       :logger #ig/reg :duct/logger}
:my-prj.handler/handle-msg {}

(defmethod ig/init-key :my-prj.handler/handle-msg [_ _]
  (fn  [ch {:keys [content-type delivery-tag type] :as meta} ^bytes payload]
    (println (format "[consumer] Received a message: %s, delivery tag: %d, content type: %s, type: %s" (String. payload "UTF-8") delivery-tag content-type type))))

Instead of :connection, you can have :connection-settings like this:

:duct.amqp.rabbitmq/langohr-consumers {:consumers [{:queue "my-queue"
                                                    :declare-queue {:durable false
                                                                    :exclusive false
                                                                    :auto-delete true
                                                                    :arguments {}}
                                                     :handler #ig/ref :my-prj.handler/handle-msg}]
                                       :connection-settings {:host "localhost"
                                       :port 5672
                                       :username "guest"
                                       :password "guest"
                                       :vhost "/"}
:my-prj.handler/handle-msg {}

Producers

:duct.amqp.rabbitmq/langohr-producers {:producers [{:name "my-producer"
                                                    :exchange "my-excahnge"
                                                    :routing-key "my-key"}]
                                       :connection #ig/ref :duct.amqp.rabbitmq/langohr}

The return value is a map whose keys are the name of the producers (left as strings) and hte values and the values are a map {:publish-fn fun :ch ch}, where fun is a function that publishes to the corresponding exchange and routing key according to langohr.basic/publish semantics, where the two first parameters are fixed according to the specified exchange and routing key in the settings.

(def publish (get-in system [:duct.amqp.rabbitmq/langohr-producers "my-producer" :publish-fn]))
(publish "Hello!" {:content-type "text/plain" :type "greetings.hi"})

Instead of :connection, you can have :connection-settings like we showed in the Consumers section.

Example

Consider an application that passes messages to itslef through RabbitMQ.

The duct config might look like this:

{:duct.amqp.rabbitmq/langohr {:host "localhost"
                             :port 5672
                             :username "guest"
                             :password "guest"
                             :vhost "/"}
:duct.amqp.rabbitmq/langohr-consumers {:consumers [{:queue "my-queue"
                                                    :declare-queue {:durable false
                                                                    :exclusive false
                                                                    :auto-delete true
                                                                    :arguments {}}
                                                     :handler #ig/ref :my-prj.handler/handle-msg}]
                                       :connection #ig/ref :duct.amqp.rabbitmq/langohr}
:my-prj.handler/handle-msg {}
:duct.amqp.rabbitmq/langohr-producers {:producers [{:name "my-producer"
                                                    :exchange "my-excahnge"
                                                    :routing-key "my-key"}]
                                       :connection #ig/ref :duct.amqp.rabbitmq/langohr}

}

And the producers and consumers might be initialized like this:

(defmethod ig/init-key :my-prj.handler/handle-msg [_ _]
  (fn  [ch {:keys [content-type delivery-tag type] :as meta} ^bytes payload]
    (println (format "[consumer] Received a message: %s, delivery tag: %d, content type: %s, type: %s" (String. payload "UTF-8") delivery-tag content-type type))))
    
(def publish (get-in system [:duct.amqp.rabbitmq/langohr-producers "my-producer" :publish-fn]))

(publish "Hello!" {:content-type "text/plain" :type "greetings.hi"})

For more information using Langohr, you can start with their getting started webpage.

License

Distributed under the Eclipse Public License either version 1.0 or (at your option) any later version.

viebel/duct-rabbitmq