Add RedisNotifier and RedisNotifierStreaming for event notification handling

Author: helightdev

krescent-redisson/src/main/kotlin/dev/helight/krescent/redisson/RedisLockProvider.kt

@@ -63,4 +63,5 @@ class RedisLockProvider(
             lock.unlockAsync(claimId).asDeferred().await()
         }
     }
-}
+}
+

krescent-redisson/src/main/kotlin/dev/helight/krescent/redisson/RedisNotifier.kt

@@ -0,0 +1,50 @@
+package dev.helight.krescent.redisson
+
+import dev.helight.krescent.event.EventMessage
+import dev.helight.krescent.source.EventPublisher
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.future.asDeferred
+import org.redisson.api.RedissonClient
+import org.redisson.api.listener.MessageListener
+import kotlin.time.Duration
+
+class RedisNotifier(
+    val client: RedissonClient,
+    val channel: String = "krescent:notifications",
+) {
+    suspend fun notify() {
+        val topic = client.getTopic(channel)
+        topic.publishAsync("").asDeferred().await()
+    }
+
+    fun wrap(publisher: EventPublisher): EventPublisher {
+        return object : EventPublisher {
+            override suspend fun publish(event: EventMessage) {
+                publisher.publish(event)
+                notify()
+            }
+
+            override suspend fun publishAll(events: List<EventMessage>) {
+                publisher.publishAll(events)
+                notify()
+            }
+        }
+    }
+
+    fun receiveTo(target: FlowCollector<Unit>): Job = CoroutineScope(Dispatchers.Default).launch {
+        val topic = client.getTopic(this@RedisNotifier.channel)
+        val listenerId = topic.addListenerAsync(String::class.java, MessageListener { _, _ ->
+            runBlocking {
+                target.emit(Unit)
+            }
+        }).asDeferred().await()
+        try {
+            delay(Duration.INFINITE)
+        } finally {
+            withContext(NonCancellable) {
+                topic.removeListenerAsync(listenerId).asDeferred().await()
+            }
+        }
+    }
+}

krescent-redisson/src/test/kotlin/dev/helight/krescent/redisson/RedisNotifier.kt

@@ -0,0 +1,56 @@
+package dev.helight.krescent.redisson
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.produceIn
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.redisson.Redisson
+import org.redisson.api.RedissonClient
+import org.redisson.config.Config
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import java.time.Duration
+import kotlin.test.Test
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+@Testcontainers
+class RedisNotifierTest {
+    companion object {
+
+        @Container
+        private val valkey = GenericContainer("valkey/valkey:latest")
+            .withExposedPorts(6379)
+            .withStartupTimeout(Duration.ofSeconds(60))
+
+        private val connectionString get() = "redis://${valkey.host}:${valkey.getMappedPort(6379)}"
+
+        fun connect(): RedissonClient = Redisson.create(Config().apply {
+            useSingleServer().setAddress(connectionString)
+        })
+    }
+
+    @Test
+    fun `Receive and publish notifications`(): Unit = runBlocking {
+        val client = connect()
+        val notifier = RedisNotifier(client)
+        val flow = MutableSharedFlow<Unit>(0)
+        val job = notifier.receiveTo(flow)
+        var hasReceived = false
+        launch {
+            val channel = flow.produceIn(this)
+            channel.receive()
+            hasReceived = true
+            channel.cancel()
+        }
+        delay(50)
+        assertFalse(hasReceived)
+        notifier.notify()
+        delay(50)
+        MutableSharedFlow<Unit>(0)
+        assertTrue(hasReceived)
+        job.cancel()
+    }
+}

krescent-redisson/src/test/kotlin/dev/helight/krescent/redisson/RedisNotifierStreaming.kt

@@ -0,0 +1,54 @@
+package dev.helight.krescent.redisson
+
+import dev.helight.krescent.source.EventPublisher
+import dev.helight.krescent.source.PollingStreamingEventSource.Companion.pollingWithNotifications
+import dev.helight.krescent.source.StreamingEventSource
+import dev.helight.krescent.source.impl.InMemoryEventStore
+import dev.helight.krescent.test.StreamingEventSourceContract
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.runBlocking
+import org.redisson.Redisson
+import org.redisson.api.RedissonClient
+import org.redisson.config.Config
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import java.time.Duration
+import java.util.*
+
+@Testcontainers
+class RedisNotifierStreamingTest : StreamingEventSourceContract {
+
+    companion object {
+
+        @Container
+        private val valkey = GenericContainer("valkey/valkey:latest")
+            .withExposedPorts(6379)
+            .withStartupTimeout(Duration.ofSeconds(60))
+
+        private val connectionString get() = "redis://${valkey.host}:${valkey.getMappedPort(6379)}"
+
+        fun connect(): RedissonClient = Redisson.create(Config().apply {
+            useSingleServer().setAddress(connectionString)
+        })
+    }
+
+    override fun execWithStreamingSource(block: suspend CoroutineScope.(StreamingEventSource, EventPublisher) -> Unit) {
+        val source = InMemoryEventStore()
+        val redis = connect()
+        val notifier = RedisNotifier(redis, "krescent:notifications:${UUID.randomUUID()}")
+        val publisher = notifier.wrap(source)
+
+        val notifications = MutableSharedFlow<Unit>(0)
+        val job = notifier.receiveTo(notifications)
+        val streaming = source.pollingWithNotifications(notifications)
+        try {
+            runBlocking {
+                this.block(streaming, publisher)
+            }
+        } finally {
+            job.cancel()
+        }
+    }
+}