Use Transformer instead of ValueTransformer in jackdaw.streams.xform

examples/xf-word-count/dev/user.clj

@@ -7,17 +7,15 @@
   which libs are actually required."
   (:gen-class)
   (:require [clojure.string :as str]
+            [clojure.tools.logging :refer [info]]
             [integrant.core :as ig]
             [integrant.repl :refer [clear go halt prep init reset reset-all]]
             [jackdaw.admin :as ja]
             [jackdaw.serdes :as js]
             [jackdaw.repl :refer :all]
             [jackdaw.streams :as j]
             [jackdaw.streams.xform :as jxf]
-            [xf-word-count :as xfwc])
-  (:import org.apache.kafka.streams.kstream.ValueTransformer
-           [org.apache.kafka.streams.state KeyValueStore Stores]
-           org.apache.kafka.streams.StreamsBuilder))
+            [xf-word-count :as xfwc]))
 
 
 (def repl-config

examples/xf-word-count/src/xf_word_count.clj

@@ -5,10 +5,7 @@
             [clojure.tools.logging :refer [info]]
             [jackdaw.serdes :as js]
             [jackdaw.streams :as j]
-            [jackdaw.streams.xform :as jxf])
-  (:import org.apache.kafka.streams.kstream.ValueTransformer
-           [org.apache.kafka.streams.state KeyValueStore Stores]
-           org.apache.kafka.streams.StreamsBuilder))
+            [jackdaw.streams.xform :as jxf]))
 
 
 (defn xf-running-total
@@ -18,17 +15,18 @@
       ([] (rf))
       ([result] (rf result))
       ([result input]
-       (let [next (as-> input %
+       (let [[k v] input
+             next (as-> v %
                     (swap-fn state #(merge-with (fnil + 0) %1 %2) %)
-                    (select-keys % (keys input))
+                    (select-keys % (keys v))
                     (map vec %))]
          (rf result next))))))
 
 (defn xf
   [state swap-fn]
   (comp
-   (map (fn [x] (str/split x #" ")))
-   (map frequencies)
+   (map (fn [[k v]] [k (str/split v #" ")]))
+   (map (fn [[k v]] [k (frequencies v)]))
    (xf-running-total state swap-fn)))
 
 
@@ -47,9 +45,9 @@
 
   ;; Evaluate the form:
   (def coll
-    ["inside every large program"
-     "is a small program"
-     "struggling to get out"])
+    [[nil "inside every large program"]
+     [nil "is a small program"]
+     [nil "struggling to get out"]])
 
   ;; Let's counts the words. Evaluate the form:
   (transduce (xf (atom {}) swap!) concat coll)

src/jackdaw/streams/xform.clj

@@ -3,7 +3,8 @@
   (:gen-class)
   (:require [jackdaw.serdes :as js]
             [jackdaw.streams :as j])
-  (:import org.apache.kafka.streams.kstream.ValueTransformer
+  (:import org.apache.kafka.streams.kstream.Transformer
+           org.apache.kafka.streams.KeyValue
            [org.apache.kafka.streams.state KeyValueStore Stores]
            org.apache.kafka.streams.StreamsBuilder))
 
@@ -32,33 +33,35 @@
     (doall (map (fn [[k v]] (.put store k v)) next))
     next))
 
-(defn value-transformer
-  "Creates an instance of org.apache.kafka.streams.kstream.ValueTransformer
-  with overrides for init, transform, and close."
+(defn add-state-store!
+  [builder]
+  "Takes a builder and adds a state store."
+  (doto ^StreamsBuilder (j/streams-builder* builder)
+    (.addStateStore (Stores/keyValueStoreBuilder
+                     (Stores/persistentKeyValueStore "transducer")
+                     (js/edn-serde)
+                     (js/edn-serde))))
+  builder)
+
+(defn transformer
+  "Takes a transducer and creates an instance of
+  org.apache.kafka.streams.kstream.Transformer with overrides for
+  init, transform, and close."
   [xf]
   (let [ctx (atom nil)]
     (reify
-      ValueTransformer
+      Transformer
       (init [_ context]
         (reset! ctx context))
-      (transform [_ v]
-        (let [^KeyValueStore store (.getStateStore @ctx "transducer")]
-          (first (into [] (xf store) [v]))))
+      (transform [_ k v]
+        (let [^KeyValueStore store (.getStateStore @ctx "transducer")
+              v (first (into [] (xf store) [[k v]]))]
+          (KeyValue/pair k v)))
       (close [_]))))
 
 (defn transduce-kstream
   [kstream xf]
   "Takes a kstream and xf and transduces the stream."
   (-> kstream
-      (j/transform-values (fn [] (value-transformer xf)) ["transducer"])
+      (j/transform (fn [] (transformer xf)) ["transducer"])
       (j/flat-map (fn [[_ v]] v))))
-
-(defn add-state-store!
-  [builder]
-  "Takes a builder and adds a state store."
-  (doto ^StreamsBuilder (j/streams-builder* builder)
-    (.addStateStore (Stores/keyValueStoreBuilder
-                     (Stores/persistentKeyValueStore "transducer")
-                     (js/edn-serde)
-                     (js/edn-serde))))
-  builder)