Merge pull request #169 from shubhang93/master

uses loop recur to retry hosts for setting HA policies

src/ziggurat/messaging/rabbitmq/cluster/producer.clj

@@ -13,30 +13,39 @@
 
 (defn get-default-ha-policy [cluster-config replica-count]
   (let [ha-mode      (get cluster-config :ha-mode "exactly")
-        ha-params    (get cluster-config :ha-params  replica-count)
+        ha-params    (get cluster-config :ha-params replica-count)
         ha-sync-mode (get cluster-config :ha-sync-mode "automatic")]
     (if (= "all" ha-mode)
       {:ha-mode "all" :ha-sync-mode ha-sync-mode}
       {:ha-mode ha-mode :ha-sync-mode ha-sync-mode :ha-params ha-params})))
 
+(defn set-ha-policy-on-host [host-endpoint username password ha-policy-body exchange-name queue-name]
+  (try
+    (binding [lh/*endpoint* host-endpoint
+              lh/*username* username
+              lh/*password* password]
+      (log/info "applying HA policies to queue: " queue-name)
+      (log/info "applying HA policies to exchange: " exchange-name)
+      (lh/set-policy "/" (str queue-name "_ha_policy")
+                     {:apply-to   "all"
+                      :pattern    (str "^" queue-name "|" exchange-name "$")
+                      :definition ha-policy-body}))
+    (catch Exception e
+      (log/error "error setting ha-policies" (.getMessage e))
+      nil)))
+
 (defn set-ha-policy [queue-name exchange-name cluster-config]
-  (let [hosts-vec (str/split (:hosts cluster-config) #",")
-        hosts     (atom hosts-vec)]
-    (with-retry {:count      (count @hosts)
-                 :wait       50
-                 :on-failure #(log/error "setting ha-policies failed " (.getMessage %))}
-      (let [host      (first @hosts)
-            _         (swap! hosts rest)
-            ha-policy (get-default-ha-policy cluster-config (get-replica-count (count hosts-vec)))]
-        (binding [lh/*endpoint* (str "http://" host ":" (get cluster-config :admin-port 15672))
-                  lh/*username* (:username cluster-config)
-                  lh/*password* (:password cluster-config)]
-          (log/info "applying HA policies to queue: " queue-name)
-          (log/info "applying HA policies to exchange: " exchange-name)
-          (lh/set-policy "/" (str queue-name "_ha_policy")
-                         {:apply-to   "all"
-                          :pattern    (str "^" queue-name "|" exchange-name "$")
-                          :definition ha-policy}))))))
+  (let [username       (:username cluster-config)
+        password       (:password cluster-config)
+        hosts-vec      (str/split (:hosts cluster-config) #",")
+        ha-policy-body (get-default-ha-policy cluster-config (get-replica-count (count hosts-vec)))]
+    (loop [hosts hosts-vec]
+      (let [host-endpoint (str "http://" (first hosts) ":" (get cluster-config :admin-port 15672))
+            resp     (set-ha-policy-on-host host-endpoint username password ha-policy-body exchange-name queue-name)
+            remaining-hosts (rest hosts)]
+        (when (and  (nil? resp)
+                    (pos? (count remaining-hosts)))
+          (recur remaining-hosts))))))
 
 (defn- declare-exchange [ch exchange]
   (le/declare ch exchange "fanout" {:durable true :auto-delete false})

test/ziggurat/messaging/rabbitmq/cluster/producer_test.clj

@@ -38,6 +38,41 @@
           ha-policy                             (rmc-prod/get-default-ha-policy rmq-cluster-conf-with-ha-mode-exactly (count (:hosts rmq-cluster-config)))]
       (is (= ha-policy expected-ha-policy)))))
 
+(deftest set-ha-policy-on-host-test
+  (testing "it should apply ha-policy on a host and return a non-nil response"
+    (let [host-endpoint "http://localhost:15672"
+          username "test"
+          password "test"
+          ha-policy-body {:foo "bar"}
+          exchange-name "test-exchange"
+          queue-name "test-queue"
+          ha-policy-name (str queue-name "_ha_policy")]
+      (with-redefs [lh/set-policy (fn [^String vhost ^String name policy]
+                                    (when (and (= vhost "/")
+                                               (= policy {:apply-to "all", :pattern "^test-queue|test-exchange$", :definition {:foo "bar"}})
+                                               (= name ha-policy-name))
+                                      {}))]
+        (is (not (nil? (rmc-prod/set-ha-policy-on-host host-endpoint username password ha-policy-body exchange-name queue-name)))))))
+
+  (testing "it should apply ha-policy on a host and return a nil response when an exception is thrown"
+    (let [host-endpoint "http://localhost:15672"
+          username "test"
+          password "test"
+          ha-policy-body {:foo "bar"}
+          exchange-name "test-exchange"
+          queue-name "test-queue"]
+      (with-redefs [lh/set-policy (fn [^String vhost ^String name policy]
+                                    (throw (Exception. "error applying ha policies")))]
+        (is (nil? (rmc-prod/set-ha-policy-on-host host-endpoint username password ha-policy-body exchange-name queue-name))))))
+
+  (testing "it should set ha-policies `hosts` number of times if an excpetion is thrown"
+    (let [call-count (atom 0)]
+      (with-redefs [lh/set-policy (fn [^String vhost ^String name policy]
+                                    (swap! call-count inc)
+                                    (throw (Exception. "error applying ha policies")))]
+        (rmc-prod/set-ha-policy "" "" {:hosts "localhost-1,localhost-2,localhost-3"})
+        (is (= @call-count 3))))))
+
 (deftest create-and-bind-queue-test
   (testing "it should create a queue,an exchange and bind the queue to the exchange but not tag the queue with a dead-letter exchange"
     (let [default-props {:durable true :auto-delete false}