KStream flatMapValues (#10)

adamko-dev · Feb 6, 2022 · 6e2c5a9 · 6e2c5a9
1 parent 0f1a8f7
commit 6e2c5a9
Showing 1 changed file with 12 additions and 0 deletions.
diff --git a/...s/kotka-streams-extensions/src/main/kotlin/dev/adamko/kotka/extensions/streams/KStream.kt b/...s/kotka-streams-extensions/src/main/kotlin/dev/adamko/kotka/extensions/streams/KStream.kt
@@ -45,6 +45,18 @@ inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.flatMap(
 }
 
 
+/** @see KStream.flatMapValues */
+inline fun <inK, inV, reified outV> KStream<inK, inV>.flatMapValues(
+  name: String? = null,
+  crossinline mapper: (key: inK, value: inV) -> Iterable<outV>
+): KStream<inK, outV> {
+  return when (name) {
+    null -> flatMapValues { k, v -> mapper(k, v) }
+    else -> flatMapValues({ k, v -> mapper(k, v) }, namedAs(name))
+  }
+}
+
+
 /** @see KStream.groupBy */
 fun <K, V, outK> KStream<K, V>.groupBy(
   grouped: Grouped<outK, V>,