riemann test

Author: danielsz

src/system/components/riemann_client.clj

@@ -1,18 +1,35 @@
 (ns system.components.riemann-client
   (:require [com.stuartsierra.component :as component]
-            [riemann.client :as r]))
+            [riemann.client :as r]
+            [clojure.tools.logging :as log])
+  (:import [io.riemann.riemann.client OverloadedException]))
 
 (defrecord RiemannClient [host port transport]
   component/Lifecycle
   (start [component]
     (let [client (case transport
                      :tcp (r/tcp-client {:host host :port port})
-                     :udp (r/udp-client {:host host :port port}))]
-      (assoc component :client client)))
+                     :udp (r/udp-client {:host host :port port}))
+          a (-> (agent {})
+                (add-watch :key (fn [_k _r _os ns] (try (deref (:promise ns) 5000 ::timeout)
+                                                       (catch Exception e (log/error (.getMessage e)))))))
+          f (fn [state client event]
+              (let [v (try
+                        (r/send-event client event)
+                        (catch OverloadedException e (log/error (.getMessage e))))]
+                (assoc state :promise v)))]
+      (assoc component :client client :send-fn (partial send a f client))))
   (stop [component]
     (if-let [client (:client component)]
       (assoc component :client (r/close! client))
       component)))
 
-(defn new-riemann-client [& {:keys [host port transport] :or {host "127.0.0.1" port 5555 transport :tcp}}]
+(defn new-riemann-client
+  "Returns a Riemann client.
+
+  `send-fn` is a function that accepts a Riemann struct, which it will
+  send in an agent threadpool (asynchronously).  The promise that the
+  Riemann `send-event` returns will be derefed in a watcher (also on
+  the threadpool), and will log errors if any occur." 
+  [& {:keys [host port transport] :or {host "127.0.0.1" port 5555 transport :tcp}}]
   (map->RiemannClient {:host host :port port :transport transport}))

test/system/components/riemann_client_test.clj

@@ -10,6 +10,17 @@
 (deftest ^:dependency riemann-client
   (let [client (component/start (new-riemann-client))]
     (is (started? client))
-    (is (not (stopped? client)))
-    (component/stop client)))
+    (component/stop client)
+    (is (stopped? client))))
 
+(deftest ^:dependency riemann-send-fn
+  (let [client (component/start (new-riemann-client))
+        send-fn (:send-fn client)]
+    (Thread/sleep 1000)
+    (is (started? client))
+    (is (= clojure.lang.Agent (type (send-fn {:service "foo" :state "ok"}))))
+    (Thread/sleep 1000)
+    (is (= io.riemann.riemann.client.MapPromise (type (:promise @(send-fn {:service "foo" :state "ok"})))))
+    (is (= riemann.codec.Msg (type (deref (:promise @(send-fn {:service "foo" :state "ok"}))))))
+    (Thread/sleep 1000)
+    (component/stop client)))