feature: Standalone kafka now hosts a grpc endpoint to communicate an…

…d understand the messages

buildSrc/build.gradle.kts

@@ -1,7 +1,9 @@
-plugins { `kotlin-dsl` }
+plugins {
+  `kotlin-dsl`
+}
 
 repositories {
-    mavenCentral()
-    google()
-    gradlePluginPortal()
+  mavenCentral()
+  google()
+  gradlePluginPortal()
 }

buildSrc/settings.gradle.kts

@@ -0,0 +1,8 @@
+rootProject.name = "buildSrc"
+dependencyResolutionManagement {
+  versionCatalogs {
+    create("libs").from(files("../gradle/libs.versions.toml"))
+  }
+}
+
+enableFeaturePreview("TYPESAFE_PROJECT_ACCESSORS")

gradle/libs.versions.toml

@@ -32,6 +32,10 @@ testcontainers = "1.19.7"
 r2dbc-mssql = "1.0.2.RELEASE"
 spotless = "6.25.0"
 detekt = "1.23.6"
+wire = "5.0.0-alpha01"
+io-grpc = "1.63.0"
+io-grpc-kotlin = "1.4.1"
+google-protobuf-kotlin = "4.26.1"
 
 [libraries]
 kotlin-stdlib-jdk8 = { module = "org.jetbrains.kotlin:kotlin-stdlib-jdk8", version.ref = "kotlin" }
@@ -108,6 +112,16 @@ spring-framework-6x-context = { module = "org.springframework:spring-context", v
 spring-framework-6x-beans = { module = "org.springframework:spring-beans", version.ref = "spring-framework-6x" }
 
 detekt-formatting = { module = "io.gitlab.arturbosch.detekt:detekt-formatting", version.ref = "detekt" }
+wire-grpc-server = { module = "com.squareup.wiregrpcserver:server", version = "1.0.0-alpha03" }
+wire-grpc-server-generator = { module = "com.squareup.wiregrpcserver:server-generator", version = "1.0.0-alpha03" }
+wire-grpc-client = { module = "com.squareup.wire:wire-grpc-client", version.ref = "wire" }
+wire-grpc-runtime = { module = "com.squareup.wire:wire-runtime", version.ref = "wire" }
+io-grpc = { module = "io.grpc:grpc-core", version.ref = "io-grpc" }
+io-grpc-stub = { module = "io.grpc:grpc-stub", version.ref = "io-grpc" }
+io-grpc-protobuf = { module = "io.grpc:grpc-protobuf", version.ref = "io-grpc" }
+io-grpc-netty = { module = "io.grpc:grpc-netty", version.ref = "io-grpc" }
+io-grpc-kotlin = { module = "io.grpc:grpc-kotlin-stub", version.ref = "io-grpc-kotlin" }
+google-protobuf-kotlin = { module = "com.google.protobuf:protobuf-kotlin", version.ref = "google-protobuf-kotlin" }
 
 [plugins]
 spring-plugin = { id = "org.jetbrains.kotlin.plugin.spring", version.ref = "kotlin" }
@@ -121,4 +135,5 @@ kover = { id = "org.jetbrains.kotlinx.kover", version.ref = "kover" }
 gitVersioning = { id = "com.palantir.git-version", version = "3.0.0" }
 spotless = { id = "com.diffplug.spotless", version.ref = "spotless" }
 detekt = { id = "io.gitlab.arturbosch.detekt", version.ref = "detekt" }
+wire = { id = "com.squareup.wire", version.ref = "wire" }
 

lib/stove-testing-e2e-kafka/build.gradle.kts

@@ -1,13 +1,58 @@
+plugins {
+  alias(libs.plugins.wire)
+}
+
 dependencies {
-    api(projects.lib.stoveTestingE2e)
-    api(libs.testcontainers.kafka)
-    implementation(libs.kafka)
-    implementation(libs.kotlinx.io.reactor.extensions)
-    implementation(libs.kotlinx.jdk8)
-    implementation(libs.kotlinx.core)
-    implementation(libs.kafkaKotlin)
+  api(projects.lib.stoveTestingE2e)
+  api(libs.testcontainers.kafka)
+  implementation(libs.kafka)
+  implementation(libs.kotlinx.io.reactor.extensions)
+  implementation(libs.kotlinx.jdk8)
+  implementation(libs.kotlinx.core)
+  implementation(libs.kafkaKotlin)
+
+  api(libs.wire.grpc.server)
+  api(libs.wire.grpc.client)
+  api(libs.wire.grpc.runtime)
+  api(libs.io.grpc)
+  api(libs.io.grpc.protobuf)
+  api(libs.io.grpc.stub)
+  api(libs.io.grpc.kotlin)
+  api(libs.io.grpc.netty)
+  api(libs.google.protobuf.kotlin)
 }
 
 dependencies {
-    testImplementation(libs.slf4j.simple)
+  testImplementation(libs.slf4j.simple)
+}
+
+buildscript {
+  dependencies {
+    classpath(libs.wire.grpc.server.generator)
+  }
+}
+
+wire {
+  sourcePath("src/main/proto")
+  kotlin {
+    rpcRole = "client"
+    rpcCallStyle = "suspending"
+    exclusive = false
+    javaInterop = true
+  }
+  kotlin {
+    custom {
+      schemaHandlerFactory = com.squareup.wire.kotlin.grpcserver.GrpcServerSchemaHandler.Factory()
+      options = mapOf(
+        "singleMethodServices" to "false",
+        "rpcCallStyle" to "suspending"
+      )
+    }
+    rpcRole = "server"
+    rpcCallStyle = "suspending"
+    exclusive = false
+    singleMethodServices = false
+    javaInterop = true
+    includes = listOf("StoveKafkaObserverService")
+  }
 }

lib/stove-testing-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/KafkaContainerOptions.kt

@@ -0,0 +1,19 @@
+package com.trendyol.stove.testing.e2e.standalone.kafka
+
+import com.trendyol.stove.testing.e2e.containers.ContainerFn
+import com.trendyol.stove.testing.e2e.containers.ContainerOptions
+import com.trendyol.stove.testing.e2e.containers.DEFAULT_REGISTRY
+import org.testcontainers.containers.KafkaContainer
+
+data class KafkaContainerOptions(
+  override val registry: String = DEFAULT_REGISTRY,
+  override val image: String = "confluentinc/cp-kafka",
+  override val tag: String = "latest",
+  val ports: List<Int> = DEFAULT_KAFKA_PORTS,
+  override val compatibleSubstitute: String? = null,
+  override val containerFn: ContainerFn<KafkaContainer> = { }
+) : ContainerOptions {
+  companion object {
+    val DEFAULT_KAFKA_PORTS = listOf(9092, 9093)
+  }
+}

lib/stove-testing-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/KafkaContext.kt

@@ -0,0 +1,40 @@
+package com.trendyol.stove.testing.e2e.standalone.kafka
+
+import arrow.core.getOrElse
+import com.trendyol.stove.testing.e2e.containers.withProvidedRegistry
+import com.trendyol.stove.testing.e2e.system.*
+import com.trendyol.stove.testing.e2e.system.abstractions.SystemNotRegisteredException
+import com.trendyol.stove.testing.e2e.system.annotations.StoveDsl
+import org.testcontainers.containers.KafkaContainer
+
+data class KafkaContext(
+  val container: KafkaContainer,
+  val options: KafkaSystemOptions
+)
+
+internal fun TestSystem.kafka(): KafkaSystem = getOrNone<KafkaSystem>().getOrElse {
+  throw SystemNotRegisteredException(KafkaSystem::class)
+}
+
+internal fun TestSystem.withKafka(options: KafkaSystemOptions = KafkaSystemOptions()): TestSystem {
+  val kafka = withProvidedRegistry(
+    options.containerOptions.imageWithTag,
+    options.containerOptions.registry,
+    options.containerOptions.compatibleSubstitute
+  ) {
+    KafkaContainer(it)
+      .withExposedPorts(*options.containerOptions.ports.toTypedArray())
+      .apply(options.containerOptions.containerFn)
+      .withReuse(this.options.keepDependenciesRunning)
+  }
+  getOrRegister(KafkaSystem(this, KafkaContext(kafka, options)))
+  return this
+}
+
+@StoveDsl
+suspend fun ValidationDsl.kafka(
+  validation: @StoveDsl suspend KafkaSystem.() -> Unit
+): Unit = validation(this.testSystem.kafka())
+
+@StoveDsl
+fun WithDsl.kafka(configure: () -> KafkaSystemOptions): TestSystem = this.testSystem.withKafka(configure())

lib/stove-testing-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/KafkaExposedConfiguration.kt

@@ -0,0 +1,7 @@
+package com.trendyol.stove.testing.e2e.standalone.kafka
+
+import com.trendyol.stove.testing.e2e.system.abstractions.ExposedConfiguration
+
+data class KafkaExposedConfiguration(
+  val bootstrapServers: String
+) : ExposedConfiguration