ser/de abstraction (#654)

Ser/de interface created and applied to the projects except kafka

examples/ktor-example/src/test/kotlin/com/stove/ktor/example/e2e/TestSystemConfig.kt → examples/ktor-example/src/test/kotlin/com/stove/ktor/example/e2e/Stove.kt

@@ -10,7 +10,7 @@ import io.kotest.core.extensions.Extension
 import io.kotest.extensions.system.SystemEnvironmentProjectListener
 import stove.ktor.example.app.objectMapperRef
 
-class TestSystemConfig : AbstractProjectConfig() {
+class Stove : AbstractProjectConfig() {
   companion object {
     init {
       stoveKafkaBridgePortDefault = "50053"

examples/spring-example/src/test/kotlin/com/stove/spring/example/e2e/TestSystemConfig.kt → examples/spring-example/src/test/kotlin/com/stove/spring/example/e2e/Stove.kt

@@ -9,7 +9,7 @@ import com.trendyol.stove.testing.e2e.wiremock.*
 import io.kotest.core.config.AbstractProjectConfig
 import org.slf4j.*
 
-class TestSystemConfig : AbstractProjectConfig() {
+class Stove : AbstractProjectConfig() {
   private val logger: Logger = LoggerFactory.getLogger("WireMockMonitor")
 
   @Suppress("LongMethod")

examples/spring-standalone-example/src/test/kotlin/com/stove/spring/standalone/example/e2e/TestSystemConfig.kt → examples/spring-standalone-example/src/test/kotlin/com/stove/spring/standalone/example/e2e/Stove.kt

@@ -10,7 +10,7 @@ import io.kotest.core.config.AbstractProjectConfig
 import org.slf4j.*
 import stove.spring.standalone.example.infrastructure.ObjectMapperConfig
 
-class TestSystemConfig : AbstractProjectConfig() {
+class Stove : AbstractProjectConfig() {
   private val logger: Logger = LoggerFactory.getLogger("WireMockMonitor")
 
   @Suppress("LongMethod")

examples/spring-streams-example/src/test/kotlin/com/stove/spring/streams/example/e2e/TestSystemConfig.kt → examples/spring-streams-example/src/test/kotlin/com/stove/spring/streams/example/e2e/Stove.kt

@@ -10,7 +10,7 @@ import com.trendyol.stove.testing.e2e.system.TestSystem.Companion.validate
 import io.kotest.core.config.AbstractProjectConfig
 import org.apache.kafka.clients.admin.NewTopic
 
-class TestSystemConfig : AbstractProjectConfig() {
+class Stove : AbstractProjectConfig() {
   @Suppress("LongMethod")
   override suspend fun beforeProject(): Unit = TestSystem()
     .also {

gradle/libs.versions.toml

@@ -61,6 +61,7 @@ kotlinx-slf4j = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-slf4j", ver
 kotlinx-knit = { module = "org.jetbrains.kotlinx:kotlinx-knit", version.ref = "knit" }
 kotlinx-io-reactor = { module = "io.projectreactor:reactor-core", version.ref = "io-reactor" }
 kotlinx-io-reactor-extensions = { module = "io.projectreactor.kotlin:reactor-kotlin-extensions", version.ref = "io-reactor-extensions" }
+kotlinx-serialization = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json-jvm", version = "1.7.3" }
 
 # Arrow
 arrow-core = { module = "io.arrow-kt:arrow-core", version.ref = "arrow" }
@@ -131,10 +132,13 @@ elastic = { module = "co.elastic.clients:elasticsearch-java", version.ref = "ela
 
 # mongo
 mongodb-kotlin-coroutine = { module = "org.mongodb:mongodb-driver-kotlin-coroutine", version.ref = "mongodb" }
+mongojack = { module = "org.mongojack:mongojack", version = "5.0.2" }
 
 # misc
 lettuce-core = { module = "io.lettuce:lettuce-core", version = "6.5.0.RELEASE" }
 logback-classic = { module = "ch.qos.logback:logback-classic", version = "1.5.12" }
+logback-core = { module = "ch.qos.logback:logback-core", version = "1.5.12" }
+log4j-slf4j2-impl = { module = "org.apache.logging.log4j:log4j-slf4j2-impl", version = "2.24.2" }
 r2dbc-mssql = { module = "io.r2dbc:r2dbc-mssql", version.ref = "r2dbc-mssql" }
 microsoft-sqlserver-jdbc = { module = "com.microsoft.sqlserver:mssql-jdbc", version = "12.8.1.jre11" }
 exposed-core = { module = "org.jetbrains.exposed:exposed-core", version.ref = "exposed" }
@@ -186,6 +190,8 @@ google-protobuf-kotlin = { module = "com.google.protobuf:protobuf-kotlin", versi
 protoc = { module = "com.google.protobuf:protoc", version.ref = "google-protobuf" }
 hoplite = { module = "com.sksamuel.hoplite:hoplite-core", version.ref = "hoplite" }
 hoplite-yaml = { module = "com.sksamuel.hoplite:hoplite-yaml", version.ref = "hoplite" }
+google-gson = { module = "com.google.code.gson:gson", version = "2.11.0" }
+
 
 caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version = "3.1.8" }
 pprint = { module = "io.exoquery:pprint-kotlin", version = "2.0.2" }
@@ -237,5 +243,6 @@ gradle-release = { id = "net.researchgate.release", version.ref = "gradle-releas
 nexusPublish = { id = "io.github.gradle-nexus.publish-plugin", version.ref = "nexusPublish" }
 testLogger = { id = "com.adarshr.test-logger", version = "4.0.0" }
 protobuf = { id = "com.google.protobuf", version = "0.9.4" }
+kotlinx-serialization = { id = "org.jetbrains.kotlin.plugin.serialization", version.ref = "kotlin" }
 
 

lib/stove-testing-e2e-couchbase/src/main/kotlin/com/trendyol/stove/testing/e2e/couchbase/CouchbaseSystem.kt

@@ -1,13 +1,12 @@
 package com.trendyol.stove.testing.e2e.couchbase
 
 import com.couchbase.client.kotlin.*
-import com.couchbase.client.kotlin.Collection
-import com.couchbase.client.kotlin.codec.*
+import com.couchbase.client.kotlin.codec.typeRef
 import com.couchbase.client.kotlin.query.*
-import com.fasterxml.jackson.databind.ObjectMapper
 import com.trendyol.stove.functional.*
 import com.trendyol.stove.testing.e2e.system.TestSystem
 import com.trendyol.stove.testing.e2e.system.abstractions.*
+import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.runBlocking
 import org.slf4j.*
 
@@ -20,10 +19,7 @@ class CouchbaseSystem internal constructor(
   internal lateinit var cluster: Cluster
 
   @PublishedApi
-  internal lateinit var collection: Collection
-
-  @PublishedApi
-  internal val objectMapper: ObjectMapper = context.options.objectMapper
+  internal lateinit var collection: com.couchbase.client.kotlin.Collection
 
   private lateinit var exposedConfiguration: CouchbaseExposedConfiguration
   private val logger: Logger = LoggerFactory.getLogger(javaClass)
@@ -59,17 +55,15 @@ class CouchbaseSystem internal constructor(
     query: String,
     assertion: (List<T>) -> Unit
   ): CouchbaseSystem {
-    val result = cluster.query(
-      statement = query,
-      metrics = false,
-      consistency = QueryScanConsistency.requestPlus()
-    ).execute().rows.map { it.contentAs<T>() }
-    val objects = result
-      .map { objectMapper.writeValueAsString(it) }
-      .map { objectMapper.readValue(it, T::class.java) }
-
-    assertion(objects)
-    return this
+    val typeRef = typeRef<T>()
+    return flow {
+      cluster.query(
+        statement = query,
+        metrics = false,
+        consistency = QueryScanConsistency.requestPlus(),
+        serializer = context.options.clusterSerDe
+      ).execute { row -> emit(context.options.clusterSerDe.deserialize(row.content, typeRef)) }
+    }.toList().also(assertion).let { this }
   }
 
   @CouchbaseDsl
@@ -174,16 +168,13 @@ class CouchbaseSystem internal constructor(
     }
   }
 
-  private fun createCluster(exposedConfiguration: CouchbaseExposedConfiguration): Cluster {
-    val jackson = JacksonJsonSerializer(objectMapper)
-    return Cluster.connect(
-      exposedConfiguration.hostsWithPort,
-      exposedConfiguration.username,
-      exposedConfiguration.password
-    ) {
-      jsonSerializer = jackson
-      transcoder = JsonTranscoder(jackson)
-    }
+  private fun createCluster(exposedConfiguration: CouchbaseExposedConfiguration): Cluster = Cluster.connect(
+    exposedConfiguration.hostsWithPort,
+    exposedConfiguration.username,
+    exposedConfiguration.password
+  ) {
+    jsonSerializer = context.options.clusterSerDe
+    transcoder = context.options.clusterTranscoder
   }
 
   companion object {

lib/stove-testing-e2e-couchbase/src/main/kotlin/com/trendyol/stove/testing/e2e/couchbase/Options.kt

@@ -2,10 +2,10 @@ package com.trendyol.stove.testing.e2e.couchbase
 
 import arrow.core.getOrElse
 import com.couchbase.client.kotlin.Cluster
-import com.fasterxml.jackson.databind.ObjectMapper
+import com.couchbase.client.kotlin.codec.*
 import com.trendyol.stove.testing.e2e.containers.*
 import com.trendyol.stove.testing.e2e.database.migrations.*
-import com.trendyol.stove.testing.e2e.serialization.StoveObjectMapper
+import com.trendyol.stove.testing.e2e.serialization.E2eObjectMapperConfig
 import com.trendyol.stove.testing.e2e.system.*
 import com.trendyol.stove.testing.e2e.system.abstractions.*
 import com.trendyol.stove.testing.e2e.system.annotations.StoveDsl
@@ -22,7 +22,8 @@ data class CouchbaseExposedConfiguration(
 data class CouchbaseSystemOptions(
   val defaultBucket: String,
   val containerOptions: CouchbaseContainerOptions = CouchbaseContainerOptions(),
-  val objectMapper: ObjectMapper = StoveObjectMapper.Default,
+  val clusterSerDe: JsonSerializer = JacksonJsonSerializer(E2eObjectMapperConfig.createObjectMapperWithDefaults()),
+  val clusterTranscoder: Transcoder = JsonTranscoder(clusterSerDe),
   override val configureExposedConfiguration: (CouchbaseExposedConfiguration) -> List<String>
 ) : SystemOptions, ConfiguresExposedConfiguration<CouchbaseExposedConfiguration> {
   internal val migrationCollection: MigrationCollection<Cluster> = MigrationCollection()

lib/stove-testing-e2e-elasticsearch/src/main/kotlin/com/trendyol/stove/testing/e2e/elasticsearch/ElasticsearchSystem.kt

@@ -5,7 +5,6 @@ import co.elastic.clients.elasticsearch.ElasticsearchClient
 import co.elastic.clients.elasticsearch._types.Refresh
 import co.elastic.clients.elasticsearch._types.query_dsl.Query
 import co.elastic.clients.elasticsearch.core.*
-import co.elastic.clients.json.jackson.JacksonJsonpMapper
 import co.elastic.clients.transport.rest_client.RestClientTransport
 import com.trendyol.stove.functional.*
 import com.trendyol.stove.testing.e2e.system.TestSystem
@@ -177,7 +176,7 @@ class ElasticsearchSystem internal constructor(
   private fun createEsClient(exposedConfiguration: ElasticSearchExposedConfiguration): ElasticsearchClient =
     context.options.clientConfigurer.restClientOverrideFn
       .getOrElse { { cfg -> restClient(cfg) } }
-      .let { RestClientTransport(it(exposedConfiguration), JacksonJsonpMapper(context.options.objectMapper)) }
+      .let { RestClientTransport(it(exposedConfiguration), context.options.jsonpMapper) }
       .let { ElasticsearchClient(it) }
 
   private fun restClient(cfg: ElasticSearchExposedConfiguration): RestClient =

lib/stove-testing-e2e-elasticsearch/src/main/kotlin/com/trendyol/stove/testing/e2e/elasticsearch/Extensions.kt

@@ -1,7 +1,6 @@
 package com.trendyol.stove.testing.e2e.elasticsearch
 
 import arrow.core.getOrElse
-import arrow.integrations.jackson.module.registerArrowModule
 import com.trendyol.stove.testing.e2e.containers.withProvidedRegistry
 import com.trendyol.stove.testing.e2e.system.*
 import com.trendyol.stove.testing.e2e.system.abstractions.SystemNotRegisteredException
@@ -13,26 +12,22 @@ import com.trendyol.stove.testing.e2e.system.annotations.StoveDsl
  * Provides an [options] class to configure the Elasticsearch container.
  * You can configure it by changing the implementation of migrator.
  */
-internal fun TestSystem.withElasticsearch(options: ElasticsearchSystemOptions): TestSystem {
-  options.objectMapper.registerArrowModule()
-
-  return withProvidedRegistry(
-    imageName = options.container.imageWithTag,
-    registry = options.container.registry,
-    compatibleSubstitute = options.container.compatibleSubstitute
-  ) { StoveElasticSearchContainer(it) }
-    .apply {
-      addExposedPorts(*options.container.exposedPorts.toIntArray())
-      withPassword(options.container.password)
-      if (options.container.disableSecurity) {
-        withEnv("xpack.security.enabled", "false")
-      }
-      withReuse(this@withElasticsearch.options.keepDependenciesRunning)
-      options.container.containerFn(this)
+internal fun TestSystem.withElasticsearch(options: ElasticsearchSystemOptions): TestSystem = withProvidedRegistry(
+  imageName = options.container.imageWithTag,
+  registry = options.container.registry,
+  compatibleSubstitute = options.container.compatibleSubstitute
+) { StoveElasticSearchContainer(it) }
+  .apply {
+    addExposedPorts(*options.container.exposedPorts.toIntArray())
+    withPassword(options.container.password)
+    if (options.container.disableSecurity) {
+      withEnv("xpack.security.enabled", "false")
     }
-    .let { getOrRegister(ElasticsearchSystem(this, ElasticsearchContext(it, options))) }
-    .let { this }
-}
+    withReuse(this@withElasticsearch.options.keepDependenciesRunning)
+    options.container.containerFn(this)
+  }
+  .let { getOrRegister(ElasticsearchSystem(this, ElasticsearchContext(it, options))) }
+  .let { this }
 
 internal fun TestSystem.elasticsearch(): ElasticsearchSystem =
   getOrNone<ElasticsearchSystem>().getOrElse {

lib/stove-testing-e2e-elasticsearch/src/main/kotlin/com/trendyol/stove/testing/e2e/elasticsearch/Options.kt

@@ -2,10 +2,11 @@ package com.trendyol.stove.testing.e2e.elasticsearch
 
 import arrow.core.*
 import co.elastic.clients.elasticsearch.ElasticsearchClient
-import com.fasterxml.jackson.databind.ObjectMapper
+import co.elastic.clients.json.JsonpMapper
+import co.elastic.clients.json.jackson.JacksonJsonpMapper
 import com.trendyol.stove.testing.e2e.containers.*
 import com.trendyol.stove.testing.e2e.database.migrations.*
-import com.trendyol.stove.testing.e2e.serialization.StoveObjectMapper
+import com.trendyol.stove.testing.e2e.serialization.*
 import com.trendyol.stove.testing.e2e.system.abstractions.*
 import com.trendyol.stove.testing.e2e.system.annotations.StoveDsl
 import org.apache.http.client.config.RequestConfig
@@ -19,7 +20,7 @@ import kotlin.time.Duration.Companion.minutes
 data class ElasticsearchSystemOptions(
   val clientConfigurer: ElasticClientConfigurer = ElasticClientConfigurer(),
   val container: ElasticContainerOptions = ElasticContainerOptions(),
-  val objectMapper: ObjectMapper = StoveObjectMapper.Default,
+  val jsonpMapper: JsonpMapper = JacksonJsonpMapper(StoveSerde.jackson.default),
   override val configureExposedConfiguration: (ElasticSearchExposedConfiguration) -> List<String>
 ) : SystemOptions, ConfiguresExposedConfiguration<ElasticSearchExposedConfiguration> {
   internal val migrationCollection: MigrationCollection<ElasticsearchClient> = MigrationCollection()

lib/stove-testing-e2e-elasticsearch/src/test/kotlin/com/trendyol/stove/testing/e2e/elasticsearch/ElasticsearchExposedCertificateTest.kt

@@ -1,9 +1,8 @@
 package com.trendyol.stove.testing.e2e.elasticsearch
 
-import arrow.integrations.jackson.module.registerArrowModule
 import com.fasterxml.jackson.module.kotlin.readValue
 import com.trendyol.stove.functional.get
-import com.trendyol.stove.testing.e2e.serialization.StoveObjectMapper
+import com.trendyol.stove.testing.e2e.serialization.*
 import com.trendyol.stove.testing.e2e.system.abstractions.StateWithProcess
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.*
@@ -25,7 +24,7 @@ class ElasticsearchExposedCertificateTest : FunSpec({
               "processId": 10496
             }
       """.trimIndent()
-    val j = StoveObjectMapper.byConfiguring { this.registerArrowModule() }
+    val j = StoveSerde.jackson.default
     val stateWithProcess = j.readValue<StateWithProcess<ElasticSearchExposedConfiguration>>(state)
     val serialize = j.writeValueAsString(stateWithProcess)
     val stateWithProcess2 = j.readValue<StateWithProcess<ElasticSearchExposedConfiguration>>(serialize)

lib/stove-testing-e2e-http/src/main/kotlin/com/trendyol/stove/testing/e2e/http/HttpSystem.kt

@@ -3,8 +3,7 @@
 package com.trendyol.stove.testing.e2e.http
 
 import arrow.core.*
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.trendyol.stove.testing.e2e.serialization.StoveObjectMapper
+import com.trendyol.stove.testing.e2e.serialization.StoveSerde
 import com.trendyol.stove.testing.e2e.system.*
 import com.trendyol.stove.testing.e2e.system.abstractions.*
 import com.trendyol.stove.testing.e2e.system.annotations.StoveDsl
@@ -17,10 +16,14 @@ import io.ktor.client.plugins.logging.*
 import io.ktor.client.request.*
 import io.ktor.client.request.forms.*
 import io.ktor.http.*
+import io.ktor.serialization.*
 import io.ktor.serialization.jackson.*
 import io.ktor.util.*
+import io.ktor.util.reflect.*
+import kotlinx.coroutines.flow.Flow
 import org.slf4j.LoggerFactory
 import java.net.http.HttpClient
+import java.nio.charset.Charset
 import kotlin.time.*
 import kotlin.time.Duration.Companion.seconds
 
@@ -29,14 +32,14 @@ private val httpSystemLogger = LoggerFactory.getLogger(HttpSystem::class.java)
 @HttpDsl
 data class HttpClientSystemOptions(
   val baseUrl: String,
-  val objectMapper: ObjectMapper = StoveObjectMapper.Default,
+  val contentConverter: ContentConverter = JacksonConverter(StoveSerde.jackson.default),
   val timeout: Duration = 30.seconds,
-  val createClient: () -> io.ktor.client.HttpClient = { jsonHttpClient(timeout, objectMapper) }
+  val createClient: () -> io.ktor.client.HttpClient = { jsonHttpClient(timeout, contentConverter) }
 ) : SystemOptions {
   companion object {
     internal fun jsonHttpClient(
       timeout: Duration,
-      objectMapper: ObjectMapper
+      converter: ContentConverter
     ): io.ktor.client.HttpClient = HttpClient(OkHttp) {
       engine {
         config {
@@ -57,9 +60,9 @@ data class HttpClientSystemOptions(
       }
 
       install(ContentNegotiation) {
-        register(ContentType.Application.Json, JacksonConverter(objectMapper))
-        register(ContentType.Application.ProblemJson, JacksonConverter(objectMapper))
-        register(ContentType.parse("application/x-ndjson"), JacksonConverter(objectMapper))
+        register(ContentType.Application.Json, converter)
+        register(ContentType.Application.ProblemJson, converter)
+        register(ContentType.parse("application/x-ndjson"), converter)
       }
 
       defaultRequest {
@@ -143,6 +146,21 @@ class HttpSystem(
     expect(it.body())
   }.let { this }
 
+  suspend inline fun <reified TExpected : Any> readJsonStream(
+    uri: String,
+    queryParams: Map<String, String> = mapOf(),
+    headers: Map<String, String> = mapOf(),
+    token: Option<String> = None,
+    expect: (Flow<TExpected>) -> Unit
+  ): HttpSystem = ktorHttpClient.prepareGet(relative(uri)) {
+    headers.forEach { (key, value) -> header(key, value) }
+    header(HttpHeaders.Accept, "application/x-ndjson")
+    queryParams.forEach { (key, value) -> parameter(key, value) }
+    token.map { header(HeaderConstants.AUTHORIZATION, HeaderConstants.bearer(it)) }
+  }.readJsonContentStream {
+    options.contentConverter.deserialize(Charset.defaultCharset(), typeInfo<TExpected>(), it) as TExpected
+  }.also { expect(it) }.let { return this }
+
   @HttpDsl
   suspend fun postAndExpectBodilessResponse(
     uri: String,