Reconnection in Langohr

gistfile1.clj

;;adapted from http://clojurerabbitmq.info/articles/error_handling.html
(ns killwabbit.core
  (:gen-class)
  (:require [langohr.core      :as rmq]
            [langohr.channel   :as lch]
            [langohr.queue     :as lq]
            [langohr.exchange  :as lx]
            [langohr.consumers :as lc]
            [langohr.basic     :as lb])
  (:import java.io.IOException
           com.rabbitmq.client.AlreadyClosedException))

(defn message-handler
  [ch {:keys [content-type delivery-tag type] :as meta} ^bytes payload]
  (println (format "[consumer] Received a message: %s"
                   (String. payload "UTF-8"))))

(defn start-consumer
  [ch ^String q]
  (lq/declare ch q :durable true :auto-delete false); :exclusive false :auto-delete false)
    (lc/subscribe ch q message-handler :auto-ack true))


(defn -main
  [& args]
  (let [conn (rmq/connect {:host "localhost"
                           :port 5672
                           :username "guest"
                           :password "guest"
                           :vhost "/"
                           :automatically-recover true 
                           :automatically-recover-topology false})
        ch   (lch/open conn)
        q    "alestorm"
        x    ""]
    (start-consumer ch q)
    (rmq/on-recovery ch (fn [ch]
                          (start-consumer ch q)))
    ;; publish messages that are routed to langohr.examples.recovery.example1.q
    ;; and demonstrate consumer recovery
    (while true
      (Thread/sleep 1000)
      (try
       ; (lb/publish ch x q "hello")
        (catch AlreadyClosedException ace
          (println "Happens when you publish while the connection is down"))
        (catch IOException ioe
          (println "Happens when an IOException is caught"))))))