fix join global

parkside-securities · Apr 7, 2021 · 6281510 · 6281510
1 parent ff13973
commit 6281510
Show file tree

Hide file tree

Showing 5 changed files with 37 additions and 18 deletions.
diff --git a/src/jackdaw/streams.clj b/src/jackdaw/streams.clj
@@ -213,8 +213,10 @@
    (p/transform-values kstream value-transformer-supplier-fn state-store-names)))
 
 (defn join-global
-  [kstream global-ktable kv-mapper joiner]
-  (p/join-global kstream global-ktable kv-mapper joiner))
+  ([kstream global-ktable kv-mapper joiner]
+   (p/join-global kstream global-ktable kv-mapper joiner))
+  ([kstream global-ktable kv-mapper joiner join-name]
+   (p/join-global kstream global-ktable kv-mapper joiner join-name)))
 
 (defn left-join-global
   ([kstream global-ktable kv-mapper joiner]

diff --git a/src/jackdaw/streams/configured.clj b/src/jackdaw/streams/configured.clj
@@ -47,8 +47,8 @@
   (global-ktable
     [_ topic-config]
     (configured-global-ktable
-      config
-      (global-ktable streams-builder topic-config)))
+     config
+     (global-ktable streams-builder topic-config)))
 
   (source-topics
     [_]
@@ -81,8 +81,8 @@
      (left-join kstream ktable value-joiner-fn)))
 
   (left-join
-   [_ ktable value-joiner-fn topic-config other-topic-config]
-   (configured-kstream
+    [_ ktable value-joiner-fn topic-config other-topic-config]
+    (configured-kstream
      config
      (left-join kstream ktable value-joiner-fn topic-config other-topic-config)))
 
@@ -250,9 +250,9 @@
   (merge
     [_ other-kstream]
     (configured-kstream
-      config
-      (merge kstream
-             other-kstream)))
+     config
+     (merge kstream
+            other-kstream)))
 
   (outer-join-windowed
     [_ other-kstream value-joiner-fn windows]
@@ -297,7 +297,7 @@
      (select-key kstream key-value-mapper-fn)))
 
   (transform
-      [this transformer-supplier-fn]
+    [this transformer-supplier-fn]
     (transform this transformer-supplier-fn []))
 
   (transform
@@ -307,7 +307,7 @@
      (transform kstream transformer-supplier-fn state-store-names)))
 
   (transform-values
-      [this value-transformer-supplier-fn]
+    [this value-transformer-supplier-fn]
     (transform-values this value-transformer-supplier-fn []))
 
   (transform-values
@@ -319,8 +319,8 @@
   (left-join-global
     [_ global-ktable kv-mapper joiner]
     (configured-kstream
-      config
-      (left-join-global kstream global-ktable kv-mapper joiner)))
+     config
+     (left-join-global kstream global-ktable kv-mapper joiner)))
 
   (left-join-global
     [_ global-ktable kv-mapper joiner join-name]
@@ -331,8 +331,14 @@
   (join-global
     [_ global-ktable kv-mapper joiner]
     (configured-kstream
-      config
-      (join-global kstream global-ktable kv-mapper joiner)))
+     config
+     (join-global kstream global-ktable kv-mapper joiner)))
+
+  (join-global
+    [_ global-ktable kv-mapper joiner]
+    (configured-kstream
+     config
+     (join-global kstream global-ktable kv-mapper joiner join-name)))
 
   (kstream* [_]
     (kstream* kstream))

diff --git a/src/jackdaw/streams/interop.clj b/src/jackdaw/streams/interop.clj
@@ -405,6 +405,15 @@
             ^KeyValueMapper (select-key-value-mapper key-value-mapper-fn)
             ^ValueJoiner (value-joiner joiner-fn))))
 
+  (join-global
+    [_ global-ktable key-value-mapper-fn joiner-fn join-name]
+    (clj-kstream
+     (.join kstream
+            ^GlobalKTable (global-ktable* global-ktable)
+            ^KeyValueMapper (select-key-value-mapper key-value-mapper-fn)
+            ^ValueJoiner (value-joiner joiner-fn)
+            (Named/as join-name))))
+
   (left-join-global
     [_ global-ktable key-value-mapper-fn joiner-fn]
     (clj-kstream
@@ -486,7 +495,7 @@
   (suppress
     [_ suppress-config]
     (clj-ktable
-       (.suppress ^KTable ktable (suppress-config->suppressed suppress-config))))
+     (.suppress ^KTable ktable (suppress-config->suppressed suppress-config))))
 
   (to-kstream
     [_]

diff --git a/src/jackdaw/streams/protocols.clj b/src/jackdaw/streams/protocols.clj
@@ -174,7 +174,8 @@
     to each value in the input stream.")
 
   (join-global
-    [kstream global-ktable kv-mapper joiner])
+    [kstream global-ktable kv-mapper joiner]
+    [kstream global-ktable kv-mapper joiner join-name])
 
   (left-join-global
     [kstream global-ktable kv-mapper joiner]

diff --git a/src/jackdaw/streams/specs.clj b/src/jackdaw/streams/specs.clj
@@ -246,7 +246,8 @@
         :args (s/cat :kstream kstream?
                      :global-ktable global-ktable?
                      :kv-mapper ifn?
-                     :joiner ifn?)
+                     :joiner ifn?
+                     :join-name (s/? string?))
         :ret kstream?)
 
 (s/fdef k/left-join-global