Change FlagsUpdateSubscriber.flow to emit strings

It seems more proper to keep usages of gRPC service definition classes
in FlagSyncService.

Author: tobyhs

src/main/kotlin/io/github/tobyhs/redisflagd/FlagSyncService.kt

@@ -12,6 +12,7 @@ import jakarta.inject.Singleton
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
 
 /**
  * Implementation of flagd's FlagSyncService that uses Redis
@@ -27,7 +28,9 @@ class FlagSyncService(
             .setFlagConfiguration(flagsRepository.getFlagConfiguration())
             .build()
         emit(firstResult)
-        emitAll(flagsUpdateSubscriber.flow)
+        emitAll(flagsUpdateSubscriber.flow.map { flagConfiguration ->
+            SyncFlagsResponse.newBuilder().setFlagConfiguration(flagConfiguration).build()
+        })
     }
 
     override suspend fun fetchAllFlags(request: FetchAllFlagsRequest): FetchAllFlagsResponse {

src/main/kotlin/io/github/tobyhs/redisflagd/data/FlagsUpdateSubscriber.kt

@@ -1,12 +1,11 @@
 package io.github.tobyhs.redisflagd.data
 
-import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse
 import kotlinx.coroutines.flow.SharedFlow
 
 /**
  * Something that subscribes to flag configuration updates
  */
 interface FlagsUpdateSubscriber {
-    /** A [SharedFlow] that emits flag configuration updates */
-    val flow: SharedFlow<SyncFlagsResponse>
+    /** A [SharedFlow] that emits updated flag configurations */
+    val flow: SharedFlow<String>
 }

src/main/kotlin/io/github/tobyhs/redisflagd/data/RedisFlagsUpdateSubscriber.kt

@@ -1,6 +1,5 @@
 package io.github.tobyhs.redisflagd.data
 
-import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse
 import io.github.tobyhs.redisflagd.di.AppCoroutineScope
 import io.lettuce.core.RedisURI
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection
@@ -22,8 +21,8 @@ class RedisFlagsUpdateSubscriber(
         @Property(name = "redis.uri") private val redisUri: String,
         @AppCoroutineScope private val appScope: CoroutineScope,
 ) : FlagsUpdateSubscriber {
-    private val _flow = MutableSharedFlow<SyncFlagsResponse>()
-    override val flow: SharedFlow<SyncFlagsResponse> = _flow
+    private val _flow = MutableSharedFlow<String>()
+    override val flow: SharedFlow<String> = _flow
 
     /**
      * Subscribes to the Redis keyspace notification channels for flag configuration updates
@@ -32,9 +31,7 @@ class RedisFlagsUpdateSubscriber(
     fun subscribe() {
         pubSubConnection.addListener {
             appScope.launch {
-                val flagConfiguration = flagsRepository.refreshFlagConfiguration()
-                val response = SyncFlagsResponse.newBuilder().setFlagConfiguration(flagConfiguration).build()
-                _flow.emit(response)
+                _flow.emit(flagsRepository.refreshFlagConfiguration())
             }
         }
         val db = RedisURI.create(redisUri).database

src/test/kotlin/io/github/tobyhs/redisflagd/FlagSyncServiceTest.kt

@@ -10,7 +10,6 @@ import io.kotest.common.ExperimentalKotest
 import io.kotest.core.spec.style.DescribeSpec
 import io.kotest.core.test.testCoroutineScheduler
 import io.kotest.matchers.collections.shouldHaveSize
-import io.kotest.matchers.shouldBe
 import io.mockk.coEvery
 import io.mockk.every
 import io.mockk.mockk
@@ -36,7 +35,7 @@ class FlagSyncServiceTest : DescribeSpec({
 
     describe("syncFlags").config(coroutineTestScope = true) {
         it("returns a flow that emits flag updates") {
-            val updateFlow = MutableSharedFlow<SyncFlagsResponse>()
+            val updateFlow = MutableSharedFlow<String>()
             every { flagsUpdateSubscriber.flow } returns updateFlow
             val responses = mutableListOf<SyncFlagsResponse>()
             val flowCollectJob = launch {
@@ -50,13 +49,12 @@ class FlagSyncServiceTest : DescribeSpec({
                         "myflag": {"state": "ENABLED", "variants": {"on": true, "off": false}, "defaultVariant": "on"}
                     }
                 }""".trimIndent()
-            val nextResponse = SyncFlagsResponse.newBuilder().setFlagConfiguration(nextFlagConfiguration).build()
-            updateFlow.emit(nextResponse)
+            updateFlow.emit(nextFlagConfiguration)
             testCoroutineScheduler.advanceUntilIdle()
 
             responses.shouldHaveSize(2)
             responses[0].flagConfiguration.shouldEqualJson(initialFlagConfiguration)
-            responses[1].shouldBe(nextResponse)
+            responses[1].flagConfiguration.shouldEqualJson(nextFlagConfiguration)
             flowCollectJob.cancel()
         }
     }

src/test/kotlin/io/github/tobyhs/redisflagd/data/RedisFlagsUpdateSubscriberTest.kt

@@ -1,6 +1,5 @@
 package io.github.tobyhs.redisflagd.data
 
-import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse
 import io.kotest.common.ExperimentalKotest
 import io.kotest.core.spec.style.DescribeSpec
 import io.kotest.core.test.testCoroutineScheduler
@@ -44,7 +43,7 @@ class RedisFlagsUpdateSubscriberTest : DescribeSpec({
             coEvery { flagsRepository.refreshFlagConfiguration() } returns flagConfiguration
 
             subscriber = RedisFlagsUpdateSubscriber(flagsRepository, pubSubConnection, redisUri, this)
-            val responses = mutableListOf<SyncFlagsResponse>()
+            val responses = mutableListOf<String>()
             val flowCollectJob = launch {
                 subscriber.flow.collect { r -> responses.add(r) }
             }
@@ -57,7 +56,7 @@ class RedisFlagsUpdateSubscriberTest : DescribeSpec({
 
             responses.shouldHaveSize(1)
             val response = responses.first()
-            response.flagConfiguration.shouldBe(flagConfiguration)
+            response.shouldBe(flagConfiguration)
             flowCollectJob.cancel()
         }
     }