Use kotlinx datetime instead of java.time

helightdev · helightdev · commit fc3c8b9fd035 · 2025-05-29T09:20:25.000+02:00
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt
@@ -1,7 +1,7 @@
 package dev.helight.krescent.checkpoint
 
 import dev.helight.krescent.event.EventMessage
-import java.time.Instant
+import kotlinx.datetime.Clock
 import kotlin.concurrent.atomics.AtomicLong
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.concurrent.atomics.incrementAndFetch
@@ -58,9 +58,9 @@ class FixedTimeRateCheckpointStrategy(
 
     override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
         if (lastCheckpoint == null) return true
-        val currentTime = Instant.now()
-        val duration = currentTime.minusMillis(lastCheckpoint.timestamp.toEpochMilli()).toEpochMilli()
-        return duration >= rate.inWholeMilliseconds
+        val currentTime = Clock.System.now()
+        val duration = currentTime - (lastCheckpoint.timestamp)
+        return duration >= rate
     }
 }
 
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointingEventSourceConsumer.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointingEventSourceConsumer.kt
@@ -5,7 +5,7 @@ import dev.helight.krescent.source.EventSourceConsumer
 import dev.helight.krescent.source.EventSourcingStrategy
 import dev.helight.krescent.source.StreamingEventSource
 import dev.helight.krescent.source.StreamingToken
-import java.time.Instant
+import kotlinx.datetime.Clock
 
 class CheckpointingEventSourceConsumer(
     val namespace: String,
@@ -73,7 +73,7 @@ class CheckpointingEventSourceConsumer(
             namespace = namespace,
             version = version,
             position = position.serialize(),
-            timestamp = Instant.now(),
+            timestamp = Clock.System.now(),
             data = bucket
         )
     }
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/StoredCheckpoint.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/StoredCheckpoint.kt
@@ -1,15 +1,13 @@
 package dev.helight.krescent.checkpoint
 
-import dev.helight.krescent.serialization.InstantSerializer
+import kotlinx.datetime.Instant
 import kotlinx.serialization.Serializable
-import java.time.Instant
 
 @Serializable
 data class StoredCheckpoint(
     val namespace: String,
     val version: String,
     val position: String,
-    @Serializable(with = InstantSerializer::class)
     val timestamp: Instant,
     val data: CheckpointBucket,
 )
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/event/Event.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/event/Event.kt
@@ -1,7 +1,7 @@
 package dev.helight.krescent.event
 
 import dev.helight.krescent.source.StreamingToken
-import java.time.Instant
+import kotlinx.datetime.Instant
 
 /**
  * Base class for all events in the Krescent framework.
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/event/EventMessage.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/event/EventMessage.kt
@@ -1,9 +1,9 @@
 package dev.helight.krescent.event
 
-import dev.helight.krescent.serialization.InstantSerializer
+import kotlinx.datetime.Clock
+import kotlinx.datetime.Instant
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.JsonElement
-import java.time.Instant
 import java.util.*
 
 /**
@@ -18,8 +18,7 @@ import java.util.*
 @Serializable
 data class EventMessage(
     val id: String = UUID.randomUUID().toString(),
-    @Serializable(with = InstantSerializer::class)
-    val timestamp: Instant = Instant.now(),
+    val timestamp: Instant = Clock.System.now(),
     val type: String,
     val payload: JsonElement,
 )
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/serialization/InstantSerializer.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/serialization/InstantSerializer.kt
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/source/StreamingEventSource.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/source/StreamingEventSource.kt
@@ -2,7 +2,7 @@ package dev.helight.krescent.source
 
 import dev.helight.krescent.event.EventMessage
 import kotlinx.coroutines.flow.Flow
-import java.time.Instant
+import kotlinx.datetime.Instant
 
 /**
  * Interface for an event source that provides a replayable stream of events.
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/source/impl/InMemoryEventStore.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/source/impl/InMemoryEventStore.kt
@@ -9,11 +9,11 @@ import dev.helight.krescent.source.SubscribingEventSource
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.datetime.Instant
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import java.nio.charset.StandardCharsets
-import java.time.Instant
 
 class InMemoryEventStore(
     private val events: MutableList<EventMessage> = mutableListOf(),
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/source/impl/MergingStreamingEventSource.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/source/impl/MergingStreamingEventSource.kt
@@ -7,10 +7,13 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.datetime.Clock
+import kotlinx.datetime.Instant
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
-import java.time.Instant
 import java.util.*
+import kotlin.time.Duration.Companion.milliseconds
+
 
 /**
  * A composite streaming event source that merges and manages multiple [StreamingEventSource] instances
@@ -81,14 +84,14 @@ class MergingStreamingEventSource(
         val buffers = keys.associateWith { LinkedList<Pair<EventMessage, StreamingToken<*>>>() }
         val cursors = initialToken.positions.toMutableMap()
         val terminated = mutableSetOf<String>()
-        val deadline = Instant.now().minusMillis(minAge)!!
+        val deadline = Clock.System.now().minus(minAge.milliseconds)
         val mutex = Mutex()
         var token: CompositeStreamingToken = initialToken
 
         suspend fun fetchBatch(key: String) {
             val cursor = cursors[key]
             val events = sources[key]?.fetchEventsAfter(cursor, batchSize)?.filter { (event, _) ->
-                event.timestamp.isBefore(deadline) // Remove events that are too recent
+                event.timestamp < deadline // Remove events that are too recent
             }?.toList() ?: emptyList()
 
             mutex.withLock {
@@ -143,7 +146,7 @@ class MergingStreamingEventSource(
         val mutex = Mutex()
         var token: CompositeStreamingToken = initialToken
         var incomingBuffer = PriorityQueue<IncomingMessage>()
-        var lastAcknowledged: Instant = Instant.ofEpochMilli(0)
+        var lastAcknowledged: Instant = Instant.fromEpochSeconds(0)
 
         suspend fun perform(): Flow<Pair<EventMessage, StreamingToken<*>>> = coroutineScope {
             channelFlow {
@@ -175,12 +178,12 @@ class MergingStreamingEventSource(
                 val outgoing = ArrayDeque<Pair<EventMessage, CompositeStreamingToken>>()
                 while (true) {
                     // Poll old enough messages, tick internal cursors and then add them to the outgoing queue
-                    val deadline = Instant.now().minusMillis(minAge)
+                    val deadline = Clock.System.now() - minAge.milliseconds
                     mutex.withLock {
-                        while (incomingBuffer.peek()?.message?.timestamp?.isBefore(deadline) ?: false) {
+                        while (incomingBuffer.peek()?.message?.timestamp?.let { it < deadline } ?: false) {
                             val oldest = incomingBuffer.poll()
                             val timestamp = oldest.message.timestamp
-                            if (timestamp.isBefore(lastAcknowledged)) {
+                            if (timestamp < lastAcknowledged) {
                                 throw OutOfOrderStreamException(
                                     sourceId = oldest.source, sourceToken = oldest.position, timestamp = timestamp
                                 )
diff --git a/krescent-core/src/test/kotlin/dev/helight/krescent/source/impl/ChronoBufferedMergeStreamTest.kt b/krescent-core/src/test/kotlin/dev/helight/krescent/source/impl/ChronoBufferedMergeStreamTest.kt
@@ -3,9 +3,9 @@ package dev.helight.krescent.source.impl
 import dev.helight.krescent.event.EventMessage
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Instant
 import kotlinx.serialization.json.JsonNull
 import org.junit.jupiter.api.Assertions
-import java.time.Instant
 import kotlin.test.Test
 
 class ChronoBufferedMergeStreamTest {
@@ -14,22 +14,22 @@ class ChronoBufferedMergeStreamTest {
     fun test() = runBlocking {
         val a = InMemoryEventStore(
             mutableListOf(
-                EventMessage(type = "1", payload = JsonNull, timestamp = Instant.ofEpochMilli(1)),
-                EventMessage(type = "5", payload = JsonNull, timestamp = Instant.ofEpochMilli(5)),
+                EventMessage(type = "1", payload = JsonNull, timestamp = Instant.fromEpochMilliseconds(1)),
+                EventMessage(type = "5", payload = JsonNull, timestamp = Instant.fromEpochMilliseconds(5)),
             )
         )
 
         val b = InMemoryEventStore(
             mutableListOf(
-                EventMessage(type = "2", payload = JsonNull, timestamp = Instant.ofEpochMilli(2)),
-                EventMessage(type = "6", payload = JsonNull, timestamp = Instant.ofEpochMilli(6)),
+                EventMessage(type = "2", payload = JsonNull, timestamp = Instant.fromEpochMilliseconds(2)),
+                EventMessage(type = "6", payload = JsonNull, timestamp = Instant.fromEpochMilliseconds(6)),
             )
         )
 
         val c = InMemoryEventStore(
             mutableListOf(
-                EventMessage(type = "3", payload = JsonNull, timestamp = Instant.ofEpochMilli(3)),
-                EventMessage(type = "4", payload = JsonNull, timestamp = Instant.ofEpochMilli(4)),
+                EventMessage(type = "3", payload = JsonNull, timestamp = Instant.fromEpochMilliseconds(3)),
+                EventMessage(type = "4", payload = JsonNull, timestamp = Instant.fromEpochMilliseconds(4)),
             )
         )
         val merged = ChronoBufferedMergeStreamEventSource.Companion.create(listOf(a, b, c)).fetchEventsAfter().toList()
diff --git a/krescent-core/src/test/kotlin/dev/helight/krescent/source/impl/InMemoryEventStoreTest.kt b/krescent-core/src/test/kotlin/dev/helight/krescent/source/impl/InMemoryEventStoreTest.kt
@@ -9,12 +9,14 @@ import kotlinx.coroutines.flow.takeWhile
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Clock
+import kotlinx.datetime.Instant
 import kotlinx.serialization.json.JsonPrimitive
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
-import java.time.Instant
 import java.util.*
+import kotlin.time.Duration.Companion.seconds
 
 class InMemoryEventStoreTest {
 
@@ -63,25 +65,25 @@ class InMemoryEventStoreTest {
 
     @Test
     fun `test getTokenAtTime returns correct token`() = runBlocking {
-        val now = Instant.now()
-        val event1 = createTestEvent(timestamp = now.minusSeconds(10))
+        val now = Clock.System.now()
+        val event1 = createTestEvent(timestamp = now - 10.seconds)
         val event2 = createTestEvent(timestamp = now)
-        val event3 = createTestEvent(timestamp = now.plusSeconds(10))
+        val event3 = createTestEvent(timestamp = now + 10.seconds)
 
         eventStore.publish(event1)
         eventStore.publish(event2)
         eventStore.publish(event3)
 
         // Test timestamp before all events
-        val tokenBefore = eventStore.getTokenAtTime(now.minusSeconds(20))
+        val tokenBefore = eventStore.getTokenAtTime(now - 20.seconds)
         Assertions.assertEquals(-1, tokenBefore.index)
 
         // Test timestamp between events
-        val tokenMiddle = eventStore.getTokenAtTime(now.minusSeconds(5))
+        val tokenMiddle = eventStore.getTokenAtTime(now - 5.seconds)
         Assertions.assertEquals(0, tokenMiddle.index)
 
         // Test timestamp after all events
-        val tokenAfter = eventStore.getTokenAtTime(now.plusSeconds(20))
+        val tokenAfter = eventStore.getTokenAtTime(now + 20.seconds)
         Assertions.assertEquals(2, tokenAfter.index)
     }
 
@@ -212,7 +214,7 @@ class InMemoryEventStoreTest {
     // Helper function to create test events
     private fun createTestEvent(
         id: String = UUID.randomUUID().toString(),
-        timestamp: Instant = Instant.now(),
+        timestamp: Instant = Clock.System.now(),
         type: String = "test-event",
         payload: JsonPrimitive = JsonPrimitive("test-payload"),
     ): EventMessage {
diff --git a/krescent-core/src/test/kotlin/dev/helight/krescent/source/impl/MergingStreamingEventSourceTest.kt b/krescent-core/src/test/kotlin/dev/helight/krescent/source/impl/MergingStreamingEventSourceTest.kt
@@ -8,25 +8,27 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Clock
+import kotlinx.datetime.Instant
 import kotlinx.serialization.json.JsonPrimitive
 import kotlinx.serialization.json.int
 import kotlinx.serialization.json.jsonPrimitive
 import org.junit.jupiter.api.Test
-import java.time.Instant
 import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.milliseconds
 
 class MergingStreamingEventSourceTest {
     @Test
     fun `Simple already ordered event join`() = runBlocking {
         val a = InMemoryEventStore(
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(0L), payload = JsonPrimitive(1)),
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(1L), payload = JsonPrimitive(2)),
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(2L), payload = JsonPrimitive(3))
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(0L), payload = JsonPrimitive(1)),
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(1L), payload = JsonPrimitive(2)),
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(2L), payload = JsonPrimitive(3))
         )
         val b = InMemoryEventStore(
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(3L), payload = JsonPrimitive(4)),
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(4L), payload = JsonPrimitive(5)),
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(5L), payload = JsonPrimitive(6))
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(3L), payload = JsonPrimitive(4)),
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(4L), payload = JsonPrimitive(5)),
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(5L), payload = JsonPrimitive(6))
         )
         val merged = MergingStreamingEventSource(
             mapOf(
@@ -41,14 +43,14 @@ class MergingStreamingEventSourceTest {
     @Test
     fun `More complex stream with resume`() = runBlocking {
         val a = InMemoryEventStore(
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(0L), payload = JsonPrimitive(1)),
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(100L), payload = JsonPrimitive(5)),
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(200L), payload = JsonPrimitive(6))
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(0L), payload = JsonPrimitive(1)),
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(100L), payload = JsonPrimitive(5)),
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(200L), payload = JsonPrimitive(6))
         )
         val b = InMemoryEventStore(
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(3L), payload = JsonPrimitive(2)),
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(4L), payload = JsonPrimitive(3)),
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(5L), payload = JsonPrimitive(4))
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(3L), payload = JsonPrimitive(2)),
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(4L), payload = JsonPrimitive(3)),
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(5L), payload = JsonPrimitive(4))
         )
         val merged = MergingStreamingEventSource(
             mapOf(
@@ -77,11 +79,11 @@ class MergingStreamingEventSourceTest {
 
     @Test
     fun `Test deadline cuts off too recent events`() = runBlocking {
-        val now = Instant.now()
+        val now = Clock.System.now()
         val a = InMemoryEventStore(
-            EventMessage(type = "a", timestamp = now.minusMillis(1000 * 8), payload = JsonPrimitive(1)),
-            EventMessage(type = "a", timestamp = now.minusMillis(1000 * 7), payload = JsonPrimitive(2)),
-            EventMessage(type = "a", timestamp = now.minusMillis(1000 * 1), payload = JsonPrimitive(3))
+            EventMessage(type = "a", timestamp = now - (1000 * 8).milliseconds, payload = JsonPrimitive(1)),
+            EventMessage(type = "a", timestamp = now - (1000 * 7).milliseconds, payload = JsonPrimitive(2)),
+            EventMessage(type = "a", timestamp = now - (1000 * 1).milliseconds, payload = JsonPrimitive(3))
         )
         val merged = MergingStreamingEventSource(
             mapOf(
@@ -98,14 +100,14 @@ class MergingStreamingEventSourceTest {
     @Test
     fun `Test batching works as expected`() = runBlocking {
         val a = InMemoryEventStore(
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(0L), payload = JsonPrimitive(1)),
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(1L), payload = JsonPrimitive(2)),
-            EventMessage(type = "a", timestamp = Instant.ofEpochSecond(2L), payload = JsonPrimitive(3))
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(0L), payload = JsonPrimitive(1)),
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(1L), payload = JsonPrimitive(2)),
+            EventMessage(type = "a", timestamp = Instant.fromEpochSeconds(2L), payload = JsonPrimitive(3))
         )
         val b = InMemoryEventStore(
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(3L), payload = JsonPrimitive(4)),
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(4L), payload = JsonPrimitive(5)),
-            EventMessage(type = "b", timestamp = Instant.ofEpochSecond(5L), payload = JsonPrimitive(6))
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(3L), payload = JsonPrimitive(4)),
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(4L), payload = JsonPrimitive(5)),
+            EventMessage(type = "b", timestamp = Instant.fromEpochSeconds(5L), payload = JsonPrimitive(6))
         )
         val merged = MergingStreamingEventSource(
             mapOf(
@@ -119,10 +121,10 @@ class MergingStreamingEventSourceTest {
 
     @Test
     fun `Check streaming works and also respects the deadline`() = runBlocking {
-        val now = Instant.now()
+        val now = Clock.System.now()
         val a = InMemoryEventStore(
-            EventMessage(type = "a", timestamp = now.minusMillis(1000 * 8), payload = JsonPrimitive(1)),
-            EventMessage(type = "a", timestamp = now.minusMillis(1000 * 7), payload = JsonPrimitive(2)),
+            EventMessage(type = "a", timestamp = now - (1000 * 8).milliseconds, payload = JsonPrimitive(1)),
+            EventMessage(type = "a", timestamp = now - (1000 * 7).milliseconds, payload = JsonPrimitive(2)),
             EventMessage(type = "a", timestamp = now, payload = JsonPrimitive(3))
         )
 
diff --git a/krescent-kurrent/src/main/kotlin/dev/helight/krescent/kurrent/KurrentMessageFactory.kt b/krescent-kurrent/src/main/kotlin/dev/helight/krescent/kurrent/KurrentMessageFactory.kt
diff --git a/krescent-mongo/src/main/kotlin/dev/helight/krescent/mongo/MongoCheckpointStorage.kt b/krescent-mongo/src/main/kotlin/dev/helight/krescent/mongo/MongoCheckpointStorage.kt
diff --git a/krescent-mongo/src/test/kotlin/dev/helight/krescent/mongo/MongoCollectionViewTest.kt b/krescent-mongo/src/test/kotlin/dev/helight/krescent/mongo/MongoCollectionViewTest.kt
