Add CheckpointingSituations test for graceful stream interruption with terminal checkpointing

Author: helightdev

krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt

@@ -23,17 +23,22 @@ object NoCheckpointStrategy : CheckpointStrategy {
 @OptIn(ExperimentalAtomicApi::class)
 class FixedEventRateCheckpointStrategy(
     private val checkpoint: Long,
+    val checkpointOnGracefulTermination: Boolean = true,
 ) : CheckpointStrategy {
 
     private val counter = AtomicLong(0)
 
+    override suspend fun tickGracefulTermination(): Boolean {
+        return checkpointOnGracefulTermination
+    }
+
     override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
         return counter.incrementAndFetch() % checkpoint == 0L
     }
 }
 
 @Suppress("unused")
-class ManualCheckpointStrategy() : CheckpointStrategy {
+class ManualCheckpointStrategy : CheckpointStrategy {
 
     private var shouldCheckpoint = false
 
@@ -54,8 +59,13 @@ class ManualCheckpointStrategy() : CheckpointStrategy {
 @Suppress("unused")
 class FixedTimeRateCheckpointStrategy(
     private val rate: Duration,
+    val checkpointOnGracefulTermination: Boolean = true,
 ) : CheckpointStrategy {
 
+    override suspend fun tickGracefulTermination(): Boolean {
+        return checkpointOnGracefulTermination
+    }
+
     override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
         if (lastCheckpoint == null) return true
         val currentTime = Clock.System.now()
@@ -66,6 +76,7 @@ class FixedTimeRateCheckpointStrategy(
 
 @Suppress("unused")
 object AlwaysCheckpointStrategy : CheckpointStrategy {
+
     override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
         return true
     }

krescent-core/src/test/kotlin/dev/helight/krescent/situations/CheckpointingSituations.kt

@@ -0,0 +1,47 @@
+package dev.helight.krescent.situations
+
+import dev.helight.krescent.bookstore.BookAddedEvent
+import dev.helight.krescent.bookstore.bookstoreEventCatalog
+import dev.helight.krescent.checkpoint.FixedEventRateCheckpointStrategy
+import dev.helight.krescent.checkpoint.impl.InMemoryCheckpointStorage
+import dev.helight.krescent.model.buildEventModel
+import dev.helight.krescent.source.impl.InMemoryEventStore
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class CheckpointingSituations {
+
+    @Test
+    fun `Graceful stream interruption with terminal checkpointing`() = runBlocking {
+        val store = InMemoryEventStore()
+        val checkpoints = InMemoryCheckpointStorage()
+        var counter = 0
+        val job = async {
+            store.buildEventModel("my-model", 1, bookstoreEventCatalog) {
+                useCheckpoints(checkpoints, FixedEventRateCheckpointStrategy(100))
+                handler {
+                    if (it is BookAddedEvent) counter++
+                }
+            }.stream()
+        }
+        store.publish(bookstoreEventCatalog.create(BookAddedEvent("1", "", "", 1.0, 1)))
+        store.publish(bookstoreEventCatalog.create(BookAddedEvent("1", "", "", 1.0, 1)))
+        delay(50)
+        job.cancelAndJoin()
+        store.publish(bookstoreEventCatalog.create(BookAddedEvent("1", "", "", 1.0, 1)))
+        assertEquals(counter, 2)
+        store.buildEventModel("my-model", 1, bookstoreEventCatalog) {
+            useCheckpoints(checkpoints, FixedEventRateCheckpointStrategy(100))
+            handler {
+                if (it is BookAddedEvent) counter++
+            }
+        }.catchup()
+        assertEquals(counter, 3)
+    }
+
+
+}