Tests

build.gradle.kts

@@ -7,6 +7,7 @@ plugins {
   `project-report`
   // `build-dashboard` // incompatible with Gradle CC
   idea
+  id("org.jetbrains.kotlinx.kover")
 }
 
 group = "dev.adamko.kotka"

buildSrc/build.gradle.kts

@@ -10,6 +10,14 @@ dependencies {
 
   // https://github.com/gradle/gradle/issues/15383#issuecomment-779893192
   implementation(files(libs.javaClass.superclass.protectionDomain.codeSource.location))
+
+  val kotlinxKoverVersion = "0.7.5"
+  implementation("org.jetbrains.kotlinx:kover-gradle-plugin:${kotlinxKoverVersion}")
+
+  val kotlinxKnitVersion = "0.5.0"
+  implementation("org.jetbrains.kotlinx:kotlinx-knit:${kotlinxKnitVersion}")
+
+  implementation("dev.jacomet.gradle.plugins:logging-capabilities:0.10.0")
 }
 
 kotlin {

buildSrc/settings.gradle.kts

@@ -12,8 +12,8 @@ dependencyResolutionManagement {
   repositoriesMode.set(RepositoriesMode.PREFER_SETTINGS)
 
   repositories {
-    gradlePluginPortal()
     mavenCentral()
+    gradlePluginPortal()
   }
 
   versionCatalogs {

buildSrc/src/main/kotlin/buildsrc/convention/kotlin-jvm.gradle.kts

@@ -3,15 +3,14 @@ package buildsrc.convention
 import org.gradle.kotlin.dsl.support.kotlinCompilerOptions
 import org.jetbrains.kotlin.gradle.dsl.KotlinJvmProjectExtension
 import org.jetbrains.kotlin.gradle.dsl.KotlinProjectExtension
-import org.jetbrains.kotlin.gradle.dsl.KotlinVersion.KOTLIN_1_5
-import org.jetbrains.kotlin.gradle.dsl.KotlinVersion.KOTLIN_1_8
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
 
 plugins {
   id("buildsrc.convention.base")
   kotlin("jvm")
   `java-library`
+  id("org.jetbrains.kotlinx.kover")
 }
 
 kotlin {

docs/basics.md

@@ -0,0 +1,174 @@
+<!--- TEST_NAME BasicsTest -->
+
+**Table of contents**
+
+<!--- TOC -->
+
+  * [Naming operators](#naming-operators)
+  * [Naming operators 2](#naming-operators-2)
+
+<!--- END -->
+
+
+<!--- INCLUDE .*\.kt
+import dev.adamko.kotka.extensions.*
+import org.apache.kafka.streams.*
+import dev.adamko.kotka.extensions.streams.*
+import org.apache.kafka.streams.kstream.*
+
+fun main() { 
+----- SUFFIX .*\.kt
+
+  println("kafkaStreamBuilder described:")
+  println(kafkaStreamBuilder.build().describe())
+  println("~~~~~~")
+  println("kotkaStreamBuilder described:")
+  println(kotkaStreamBuilder.build().describe())
+  println("~~~~~~")
+}
+-->
+
+### Naming operators
+
+https://kafka.apache.org/documentation/streams/developer-guide/dsl-topology-naming.html
+
+* `Grouped`
+* `StreamJoined`
+* `Joined`
+* `StreamJoined`
+* `Materialized`
+* `Named`
+
+<table>
+<tr>
+<th>Plain Kafka Streams</th>
+<th>Kotka Streams</th>
+</tr>
+<tr>
+<td>
+
+```kotlin
+val kafkaStreamBuilder = StreamsBuilder()
+val kafkaTransactions: KStream<String, String> = kafkaStreamBuilder
+  .stream("input", Consumed.`as`("Customer_transactions_input_topic"))
+
+kafkaTransactions
+  .filter(
+    { _, value -> value != "invalid_txn" },
+    Named.`as`("filter_out_invalid_txns")
+  )
+  .mapValues(
+    { _, value -> value.take(6) },
+    Named.`as`("Map_values_to_first_6_characters")
+  )
+  .to("output", Produced.`as`("Mapped_transactions_output_topic"))
+```
+
+</td>
+<td>
+
+```kotlin
+val kotkaStreamBuilder = StreamsBuilder()
+val kotkaTransactions: KStream<String, String> = kotkaStreamBuilder
+  // no need for backticks 
+  .stream("input", consumedAs("Customer_transactions_input_topic"))
+
+kotkaTransactions
+  // tasks can be named directly using a string, 
+  // and the lambda expression can be placed outside the parentheses 
+  .filter(name = "filter_out_invalid_txns") { _, value ->
+    value != "invalid_txn"
+  }
+  .mapValues(name = "Map_values_to_first_6_characters") { _, value ->
+    value.take(6)
+  }
+  .to("output", producedAs("Mapped_transactions_output_topic"))
+```
+
+
+</td>
+</tr>
+</table>
+
+> You can get the full code [here](./code/example/example-basics-naming-operators-01.kt).
+
+Both examples produce the same topology:
+
+```text
+Topologies:
+   Sub-topology: 0
+    Source: Customer_transactions_input_topic (topics: [input])
+      --> filter_out_invalid_txns
+    Processor: filter_out_invalid_txns (stores: [])
+      --> Map_values_to_first_6_characters
+      <-- Customer_transactions_input_topic
+    Processor: Map_values_to_first_6_characters (stores: [])
+      --> Mapped_transactions_output_topic
+      <-- filter_out_invalid_txns
+    Sink: Mapped_transactions_output_topic (topic: output)
+      <-- Map_values_to_first_6_characters
+```
+
+<!--- TEST -->
+
+### Naming operators 2
+
+
+
+```kotlin
+val kafkaStreamBuilder = StreamsBuilder()
+val kafkaTransactions: KStream<String, String> = kafkaStreamBuilder
+  .stream("input", Consumed.`as`("Customer_transactions_input_topic"))
+
+kafkaTransactions
+  .filter(
+    { _, v -> v != "invalid_txn" },
+    Named.`as`("filter_out_invalid_txns")
+  )
+  .mapValues(
+    { _, v -> v.take(6) },
+    Named.`as`("Map_values_to_first_6_characters")
+  )
+  .to("output", Produced.`as`("Mapped_transactions_output_topic"))
+```
+
+
+```kotlin
+val kotkaStreamBuilder = StreamsBuilder()
+val kotkaTransactions: KStream<String, String> = kotkaStreamBuilder
+  // no need for backticks 
+  .stream("input", consumedAs("Customer_transactions_input_topic"))
+
+kotkaTransactions
+  // tasks can be named directly using a string, 
+  // and the lambda expression can be placed outside the parentheses 
+  .filter(name = "filter_out_invalid_txns") { _, value ->
+    value != "invalid_txn"
+  }
+  .mapValues(name = "Map_values_to_first_6_characters") { _, value ->
+    value.take(6)
+  }
+  .to("output", producedAs("Mapped_transactions_output_topic"))
+```
+
+
+> You can get the full code [here](./code/example/example-basics-naming-operators-02.kt).
+
+Both examples produce the same topology:
+
+```text
+Topologies:
+   Sub-topology: 0
+    Source: Customer_transactions_input_topic (topics: [input])
+      --> filter_out_invalid_txns
+    Processor: filter_out_invalid_txns (stores: [])
+      --> Map_values_to_first_6_characters
+      <-- Customer_transactions_input_topic
+    Processor: Map_values_to_first_6_characters (stores: [])
+      --> Mapped_transactions_output_topic
+      <-- filter_out_invalid_txns
+    Sink: Mapped_transactions_output_topic (topic: output)
+      <-- Map_values_to_first_6_characters
+```
+
+<!--- TEST -->

docs/code/build.gradle.kts

@@ -0,0 +1,57 @@
+plugins {
+  buildsrc.convention.`kotlin-jvm`
+  id("org.jetbrains.kotlinx.knit")
+}
+
+
+dependencies {
+  implementation(projects.modules.kotkaStreamsExtensions)
+
+  testImplementation(kotlin("test"))
+
+  implementation(libs.slf4j.api)
+  implementation(libs.slf4j.simple)
+
+  implementation(libs.kotlinx.knit)
+  testImplementation(libs.kotlinx.knitTest)
+
+  testImplementation(platform(libs.kotest.bom))
+  testImplementation(libs.kotest.assertionsCore)
+}
+
+
+sourceSets.test {
+  java.srcDirs(
+    layout.projectDirectory.dir("example"),
+    layout.projectDirectory.dir("test"),
+  )
+}
+
+knit {
+  val docsDir = rootProject.layout.projectDirectory.dir("docs")
+  rootDir = docsDir.asFile
+  files = project.fileTree(docsDir) {
+    include("*.md")
+  }
+}
+
+tasks.test {
+  dependsOn(tasks.knit)
+  dependsOn(tasks.processResources)
+//  finalizedBy(tasks.knitCheck)
+}
+
+tasks.compileKotlin { mustRunAfter(tasks.knit) }
+
+tasks.knitCheck {
+  dependsOn(tasks.test)
+}
+
+tasks.knit {
+  inputs.files(
+    layout.projectDirectory.files(
+      "knit-tests.ftl",
+      "knit-includes.ftl",
+    )
+  ).withPropertyName("knitTemplates")
+}

docs/code/example/example-basics-naming-operators-01.kt

@@ -0,0 +1,49 @@
+// This file was automatically generated from basics.md by Knit tool. Do not edit.
+@file:Suppress("PackageDirectoryMismatch", "unused")
+package dev.adamko.kotka.example.exampleBasicsNamingOperators01
+
+import dev.adamko.kotka.extensions.*
+import org.apache.kafka.streams.*
+import dev.adamko.kotka.extensions.streams.*
+import org.apache.kafka.streams.kstream.*
+
+fun main() { 
+
+val kafkaStreamBuilder = StreamsBuilder()
+val kafkaTransactions: KStream<String, String> = kafkaStreamBuilder
+  .stream("input", Consumed.`as`("Customer_transactions_input_topic"))
+
+kafkaTransactions
+  .filter(
+    { _, value -> value != "invalid_txn" },
+    Named.`as`("filter_out_invalid_txns")
+  )
+  .mapValues(
+    { _, value -> value.take(6) },
+    Named.`as`("Map_values_to_first_6_characters")
+  )
+  .to("output", Produced.`as`("Mapped_transactions_output_topic"))
+
+val kotkaStreamBuilder = StreamsBuilder()
+val kotkaTransactions: KStream<String, String> = kotkaStreamBuilder
+  // no need for backticks 
+  .stream("input", consumedAs("Customer_transactions_input_topic"))
+
+kotkaTransactions
+  // tasks can be named directly using a string, 
+  // and the lambda expression can be placed outside the parentheses 
+  .filter(name = "filter_out_invalid_txns") { _, value ->
+    value != "invalid_txn"
+  }
+  .mapValues(name = "Map_values_to_first_6_characters") { _, value ->
+    value.take(6)
+  }
+  .to("output", producedAs("Mapped_transactions_output_topic"))
+
+  println("kafkaStreamBuilder described:")
+  println(kafkaStreamBuilder.build().describe())
+  println("~~~~~~")
+  println("kotkaStreamBuilder described:")
+  println(kotkaStreamBuilder.build().describe())
+  println("~~~~~~")
+}

docs/code/example/example-basics-naming-operators-02.kt

@@ -0,0 +1,49 @@
+// This file was automatically generated from basics.md by Knit tool. Do not edit.
+@file:Suppress("PackageDirectoryMismatch", "unused")
+package dev.adamko.kotka.example.exampleBasicsNamingOperators02
+
+import dev.adamko.kotka.extensions.*
+import org.apache.kafka.streams.*
+import dev.adamko.kotka.extensions.streams.*
+import org.apache.kafka.streams.kstream.*
+
+fun main() { 
+
+val kafkaStreamBuilder = StreamsBuilder()
+val kafkaTransactions: KStream<String, String> = kafkaStreamBuilder
+  .stream("input", Consumed.`as`("Customer_transactions_input_topic"))
+
+kafkaTransactions
+  .filter(
+    { _, v -> v != "invalid_txn" },
+    Named.`as`("filter_out_invalid_txns")
+  )
+  .mapValues(
+    { _, v -> v.take(6) },
+    Named.`as`("Map_values_to_first_6_characters")
+  )
+  .to("output", Produced.`as`("Mapped_transactions_output_topic"))
+
+val kotkaStreamBuilder = StreamsBuilder()
+val kotkaTransactions: KStream<String, String> = kotkaStreamBuilder
+  // no need for backticks 
+  .stream("input", consumedAs("Customer_transactions_input_topic"))
+
+kotkaTransactions
+  // tasks can be named directly using a string, 
+  // and the lambda expression can be placed outside the parentheses 
+  .filter(name = "filter_out_invalid_txns") { _, value ->
+    value != "invalid_txn"
+  }
+  .mapValues(name = "Map_values_to_first_6_characters") { _, value ->
+    value.take(6)
+  }
+  .to("output", producedAs("Mapped_transactions_output_topic"))
+
+  println("kafkaStreamBuilder described:")
+  println(kafkaStreamBuilder.build().describe())
+  println("~~~~~~")
+  println("kotkaStreamBuilder described:")
+  println(kotkaStreamBuilder.build().describe())
+  println("~~~~~~")
+}

docs/code/knit-include.ftl

@@ -0,0 +1,6 @@
+<#-- @ftlvariable name="knit.name" type="String" -->
+<#-- @ftlvariable name="knit.package" type="String" -->
+<#-- @ftlvariable name="file.name" type="String" -->
+// This file was automatically generated from ${file.name} by Knit tool. Do not edit.
+@file:Suppress("PackageDirectoryMismatch", "unused")
+package ${knit.package}.${knit.name}