Use function to infer redis key from event

Author: nukemberg

README.md

@@ -42,7 +42,8 @@ Options:
 - `:conn-spec` - [Carmine](https://github.com/ptaoussanis/carmine) redis connection spec. See Carmine docs for more info
 - `:pool-spec` - Carmine connection pool spec, basically apache-commons pool options
 - `:encoder` - A function to convert Riemann event map to a string. Defaults to `cheshire/generate-string` (JSON serialization), see the source for more info.
-- `:key` - The name of the Redis key to output events to. Defaults to `"riemann-events"`
+- `:key` - The Redis key to output events to. Defaults to `"riemann-events"`
+- `:key-fn` - A function to compute the Redis key from event; Overrides `:key`  
 
 ### Flapjack:
 

src/riemann_redis_output/core.clj

@@ -51,13 +51,15 @@
   :conn-spec - Carmine spec, e.g. {:host \"redisserver\" :port 6379 :db 13}, see Carmine for details
   :pool-spec - Carmin pool spec (Apache commons pool options). See Carmine for details
   :key - Redis key to push events to
+  :key-fn - A function to compute the Redis key from event; overrides `:key`. Signature (fn [event]) -> String
   :encoder - the encoding function to convert event map to a string. Signature: (fn [event]) -> String. Defaults to cheshire/generate-string
   "
   ([] (sink {}))
   ([opts]
-   (let [{:keys [conn-spec key encoder pool-spec]} (merge default-opts opts)]
+   (let [{:keys [conn-spec key key-fn encoder pool-spec]} (merge default-opts opts)
+         key-fn (or key-fn (constantly key))]
      (fn [events]
-       (write {:pool pool-spec :spec conn-spec} key encoder
-              (if (sequential? events)
-                events
-                [events]))))))
+       (let [events (if (sequential? events) events [events])
+             events-by-keys (group-by key-fn events)]
+         (doseq [[key events] events-by-keys]
+           (write {:pool pool-spec :spec conn-spec} key encoder events)))))))

test/riemann_redis_output/core_test.clj

@@ -3,7 +3,7 @@
             [cheshire.core :refer [parse-string]]
             [riemann-redis-output.core :refer [flapjack-encoder]]))
 
-(defn- rand-str [len]
+(defn rand-str [len]
   (apply str (take len (repeatedly #(char (+ (rand 26) 65))))))
 
 (deftest flapjack

test/riemann_redis_output/integration_test.clj

@@ -7,7 +7,8 @@
             [clj-test-containers.core :as tc]
             [riemann-redis-output.core :refer [sink]]
             [cheshire.core :as json]
-            [taoensso.carmine :as car]))
+            [taoensso.carmine :as car]
+            [riemann-redis-output.core-test :refer [rand-str]]))
 
 (def ^:dynamic redis-port 6379)
 (defn- redis-fixture [f]
@@ -79,3 +80,15 @@
                                 (car/lpop "test-byte-array"))
           redis-event (json/parse-string redis-event true)]
       (is (= "tested" (:state redis-event))))))
+
+(deftest output-with-key-field
+  (let [key-field :redis-key
+        event {:time 0 :state "ok" :service "test service" :host "localhost" :description (rand-str 10) :redis-key (rand-str 10)}
+        redis-out (sink {:conn-spec {:port redis-port} :key-fn key-field})]
+    (with-server (streams/sdo redis-out)
+      (send-events [event])
+      (Thread/sleep 50))
+    (let [redis-event (car/wcar {:pool :none :spec {:host "localhost" :port redis-port}}
+                                (car/lpop (:redis-key event)))
+          redis-event (json/parse-string redis-event true)]
+      (is (= (:description event) (:description redis-event))))))