Compare commits
No commits in common. "master" and "v1.0-rc27" have entirely different histories.
29
.github/workflows/test.yml
vendored
29
.github/workflows/test.yml
vendored
@ -1,29 +0,0 @@
|
||||
name: Run Unit Tests
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "**"
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up JDK 21
|
||||
uses: actions/setup-java@v4
|
||||
with:
|
||||
java-version: '21'
|
||||
distribution: 'zulu'
|
||||
|
||||
- name: Setup Gradle
|
||||
uses: gradle/gradle-build-action@v3
|
||||
|
||||
- name: Make gradlew executable
|
||||
run: chmod +x ./gradlew
|
||||
|
||||
- name: Run unit tests
|
||||
run: ./gradlew test --stacktrace
|
||||
@ -1,2 +1 @@
|
||||
kotlin.code.style=official
|
||||
org.gradle.parallel=false
|
||||
|
||||
@ -1,14 +1,12 @@
|
||||
package no.iktdev.eventi
|
||||
|
||||
import java.time.Clock
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
|
||||
object MyTime {
|
||||
private val clock: Clock = Clock.systemUTC()
|
||||
|
||||
@JvmStatic
|
||||
fun utcNow(): Instant =
|
||||
Instant.now(clock)
|
||||
|
||||
fun UtcNow(): LocalDateTime =
|
||||
LocalDateTime.now(clock)
|
||||
}
|
||||
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
package no.iktdev.eventi
|
||||
|
||||
import com.google.gson.Gson
|
||||
import com.google.gson.GsonBuilder
|
||||
import com.google.gson.JsonDeserializationContext
|
||||
import com.google.gson.JsonDeserializer
|
||||
@ -16,14 +15,13 @@ import no.iktdev.eventi.models.store.PersistedTask
|
||||
import no.iktdev.eventi.models.store.TaskStatus
|
||||
import no.iktdev.eventi.tasks.TaskTypeRegistry
|
||||
import java.lang.reflect.Type
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import java.time.format.DateTimeFormatter
|
||||
|
||||
object ZDS {
|
||||
val gson = WGson.gson
|
||||
|
||||
fun Event.toPersisted(id: Long, persistedAt: Instant = MyTime.utcNow()): PersistedEvent? {
|
||||
fun Event.toPersisted(id: Long, persistedAt: LocalDateTime = MyTime.UtcNow()): PersistedEvent? {
|
||||
val payloadJson = gson.toJson(this)
|
||||
val eventName = this::class.simpleName ?: run {
|
||||
throw IllegalStateException("Missing class name for event: $this")
|
||||
@ -49,7 +47,7 @@ object ZDS {
|
||||
return gson.fromJson(data, clazz)
|
||||
}
|
||||
|
||||
fun Task.toPersisted(id: Long, status: TaskStatus = TaskStatus.Pending, persistedAt: Instant = MyTime.utcNow()): PersistedTask? {
|
||||
fun Task.toPersisted(id: Long, status: TaskStatus = TaskStatus.Pending, persistedAt: LocalDateTime = MyTime.UtcNow()): PersistedTask? {
|
||||
val payloadJson = gson.toJson(this)
|
||||
val taskName = this::class.simpleName ?: run {
|
||||
throw IllegalStateException("Missing class name for task: $this")
|
||||
@ -82,47 +80,26 @@ object ZDS {
|
||||
|
||||
object WGson {
|
||||
val gson = GsonBuilder()
|
||||
.registerTypeAdapter(Instant::class.java, InstantAdapter())
|
||||
// hvis du fortsatt har LocalDateTime et sted:
|
||||
.registerTypeAdapter(LocalDateTime::class.java, LocalDateTimeAdapter())
|
||||
.create()
|
||||
|
||||
fun toJson(data: Any?): String =
|
||||
gson.toJson(data)
|
||||
|
||||
class InstantAdapter : JsonSerializer<Instant>, JsonDeserializer<Instant> {
|
||||
override fun serialize(
|
||||
src: Instant,
|
||||
typeOfSrc: Type,
|
||||
context: JsonSerializationContext
|
||||
): JsonElement =
|
||||
JsonPrimitive(src.toString()) // ISO-8601, UTC
|
||||
|
||||
override fun deserialize(
|
||||
json: JsonElement,
|
||||
typeOfT: Type,
|
||||
context: JsonDeserializationContext
|
||||
): Instant =
|
||||
Instant.parse(json.asString)
|
||||
fun toJson(data: Any?): String {
|
||||
return gson.toJson(data)
|
||||
}
|
||||
|
||||
class LocalDateTimeAdapter : JsonSerializer<LocalDateTime>, JsonDeserializer<LocalDateTime> {
|
||||
private val formatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME
|
||||
|
||||
override fun serialize(
|
||||
src: LocalDateTime,
|
||||
typeOfSrc: Type,
|
||||
context: JsonSerializationContext
|
||||
): JsonElement =
|
||||
JsonPrimitive(src.format(formatter))
|
||||
src: LocalDateTime, typeOfSrc: Type, context: JsonSerializationContext
|
||||
): JsonElement {
|
||||
return JsonPrimitive(src.format(formatter))
|
||||
}
|
||||
|
||||
override fun deserialize(
|
||||
json: JsonElement,
|
||||
typeOfT: Type,
|
||||
context: JsonDeserializationContext
|
||||
): LocalDateTime =
|
||||
LocalDateTime.parse(json.asString, formatter)
|
||||
json: JsonElement, typeOfT: Type, context: JsonDeserializationContext
|
||||
): LocalDateTime {
|
||||
return LocalDateTime.parse(json.asString, formatter)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -2,7 +2,6 @@ package no.iktdev.eventi.events
|
||||
|
||||
import no.iktdev.eventi.models.DeleteEvent
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.SignalEvent
|
||||
import no.iktdev.eventi.stores.EventStore
|
||||
import java.util.UUID
|
||||
|
||||
@ -12,18 +11,12 @@ open class EventDispatcher(val eventStore: EventStore) {
|
||||
val derivedFromIds = events.mapNotNull { it.metadata.derivedFromId }.flatten().toSet()
|
||||
val deletedEventIds = events.filterIsInstance<DeleteEvent>().map { it.deletedEventId }
|
||||
val candidates = events
|
||||
.filterNot { it is SignalEvent }
|
||||
.filter { it.eventId !in derivedFromIds }
|
||||
.filter { it.eventId !in deletedEventIds }
|
||||
|
||||
val effectiveHistory = events
|
||||
.filter { it.eventId !in deletedEventIds } // fjern slettede events
|
||||
.filterNot { it is DeleteEvent } // fjern selve delete-eventet
|
||||
|
||||
|
||||
EventListenerRegistry.getListeners().forEach { listener ->
|
||||
for (candidate in candidates) {
|
||||
val result = listener.onEvent(candidate, effectiveHistory)
|
||||
val result = listener.onEvent(candidate, events)
|
||||
if (result != null) {
|
||||
eventStore.persist(result)
|
||||
}
|
||||
|
||||
@ -6,32 +6,19 @@ import no.iktdev.eventi.MyTime
|
||||
import no.iktdev.eventi.ZDS.toEvent
|
||||
import no.iktdev.eventi.stores.EventStore
|
||||
import java.time.Duration
|
||||
import java.time.Instant
|
||||
import java.util.UUID
|
||||
import java.time.LocalDateTime
|
||||
import kotlin.collections.iterator
|
||||
|
||||
abstract class EventPollerImplementation(
|
||||
private val eventStore: EventStore,
|
||||
private val dispatchQueue: SequenceDispatchQueue,
|
||||
private val dispatcher: EventDispatcher
|
||||
) {
|
||||
private val log = KotlinLogging.logger {}
|
||||
|
||||
/**
|
||||
* Per-reference watermark:
|
||||
* - first = last seen persistedAt
|
||||
* - second = last seen persistedId
|
||||
*/
|
||||
protected val refWatermark = mutableMapOf<UUID, Pair<Instant, Long>>()
|
||||
|
||||
/**
|
||||
* Global scan hint (timestamp only).
|
||||
* Used to avoid scanning entire table every time.
|
||||
*/
|
||||
var lastSeenTime: Instant = Instant.EPOCH
|
||||
|
||||
var lastSeenTime: LocalDateTime = LocalDateTime.of(1970, 1, 1, 0, 0)
|
||||
open var backoff = Duration.ofSeconds(2)
|
||||
protected set
|
||||
private val maxBackoff = Duration.ofMinutes(1)
|
||||
private val log = KotlinLogging.logger {}
|
||||
|
||||
open suspend fun start() {
|
||||
log.info { "EventPoller starting with initial backoff=$backoff" }
|
||||
@ -39,7 +26,7 @@ abstract class EventPollerImplementation(
|
||||
try {
|
||||
pollOnce()
|
||||
} catch (e: Exception) {
|
||||
log.error(e) { "Error in poller loop" }
|
||||
e.printStackTrace()
|
||||
delay(backoff.toMillis())
|
||||
backoff = backoff.multipliedBy(2).coerceAtMost(maxBackoff)
|
||||
}
|
||||
@ -47,92 +34,46 @@ abstract class EventPollerImplementation(
|
||||
}
|
||||
|
||||
suspend fun pollOnce() {
|
||||
val pollStartedAt = MyTime.utcNow()
|
||||
log.debug { "🔍 Polling for new events" }
|
||||
|
||||
// Determine global scan start
|
||||
val minRefTs = refWatermark.values.minOfOrNull { it.first }
|
||||
val scanFrom = when (minRefTs) {
|
||||
null -> lastSeenTime
|
||||
else -> maxOf(lastSeenTime, minRefTs)
|
||||
}
|
||||
|
||||
val newPersisted = eventStore.getPersistedEventsAfter(scanFrom)
|
||||
val pollStartedAt = MyTime.UtcNow()
|
||||
log.debug { "Polling for new events" }
|
||||
val newPersisted = eventStore.getPersistedEventsAfter(lastSeenTime)
|
||||
|
||||
if (newPersisted.isEmpty()) {
|
||||
log.debug { "😴 No new events found. Backing off for $backoff" }
|
||||
log.debug { "No new events found. Backing off for $backoff" }
|
||||
delay(backoff.toMillis())
|
||||
backoff = backoff.multipliedBy(2).coerceAtMost(maxBackoff)
|
||||
return
|
||||
}
|
||||
|
||||
// Reset backoff
|
||||
backoff = Duration.ofSeconds(2)
|
||||
log.debug { "📬 Found ${newPersisted.size} new events after $scanFrom" }
|
||||
|
||||
val grouped = newPersisted.groupBy { it.referenceId }
|
||||
var anyProcessed = false
|
||||
|
||||
// Track highest persistedAt seen globally this round
|
||||
val maxPersistedThisRound = newPersisted.maxOf { it.persistedAt }
|
||||
// Samle persistedAt KUN for referanser vi faktisk dispatch’et
|
||||
val processedTimes = mutableListOf<LocalDateTime>()
|
||||
|
||||
for ((ref, eventsForRef) in grouped) {
|
||||
val (refSeenAt, refSeenId) = refWatermark[ref] ?: (Instant.EPOCH to 0L)
|
||||
|
||||
// Filter new events using (timestamp, id) ordering
|
||||
val newForRef = eventsForRef.filter { ev ->
|
||||
ev.persistedAt > refSeenAt ||
|
||||
(ev.persistedAt == refSeenAt && ev.id > refSeenId)
|
||||
}
|
||||
|
||||
if (newForRef.isEmpty()) {
|
||||
log.debug { "🧊 No new events for $ref since ($refSeenAt, id=$refSeenId)" }
|
||||
for ((referenceId, _) in grouped) {
|
||||
if (dispatchQueue.isProcessing(referenceId)) {
|
||||
log.debug { "Skipping dispatch for $referenceId as it is already being processed" }
|
||||
continue
|
||||
}
|
||||
|
||||
// If ref is busy, skip dispatch
|
||||
if (dispatchQueue.isProcessing(ref)) {
|
||||
log.debug { "⏳ $ref is busy — deferring ${newForRef.size} events" }
|
||||
continue
|
||||
}
|
||||
|
||||
// Fetch full sequence for dispatch
|
||||
val fullLog = eventStore.getPersistedEventsFor(ref)
|
||||
val fullLog = eventStore.getPersistedEventsFor(referenceId)
|
||||
val events = fullLog.mapNotNull { it.toEvent() }
|
||||
|
||||
log.debug { "🚀 Dispatching ${events.size} events for $ref" }
|
||||
dispatchQueue.dispatch(ref, events, dispatcher)
|
||||
|
||||
// Update watermark for this reference
|
||||
val maxEvent = newForRef.maxWith(
|
||||
compareBy({ it.persistedAt }, { it.id })
|
||||
)
|
||||
|
||||
val newWatermarkAt = minOf(pollStartedAt, maxEvent.persistedAt)
|
||||
val newWatermarkId = maxEvent.id
|
||||
|
||||
refWatermark[ref] = newWatermarkAt to newWatermarkId
|
||||
anyProcessed = true
|
||||
|
||||
log.debug { "⏩ Updated watermark for $ref → ($newWatermarkAt, id=$newWatermarkId)" }
|
||||
processedTimes += fullLog.map { it.persistedAt }
|
||||
dispatchQueue.dispatch(referenceId, events, dispatcher)
|
||||
}
|
||||
|
||||
// Update global scan hint
|
||||
val newLastSeen = maxOf(
|
||||
lastSeenTime,
|
||||
maxPersistedThisRound.plusNanos(1)
|
||||
)
|
||||
|
||||
if (anyProcessed) {
|
||||
val minRef = refWatermark.values.minOfOrNull { it.first }
|
||||
lastSeenTime = when (minRef) {
|
||||
null -> newLastSeen
|
||||
else -> maxOf(newLastSeen, minRef)
|
||||
}
|
||||
log.debug { "📉 Global scanFrom updated → $lastSeenTime (anyProcessed=true)" }
|
||||
} else {
|
||||
if (processedTimes.isNotEmpty()) {
|
||||
val maxPersistedAt = processedTimes.max()
|
||||
val newLastSeen = minOf(pollStartedAt, maxPersistedAt).plusNanos(1)
|
||||
log.debug { "Updating lastSeenTime from $lastSeenTime to $newLastSeen" }
|
||||
lastSeenTime = newLastSeen
|
||||
log.debug { "🔁 No refs processed — advancing global scanFrom to $lastSeenTime" }
|
||||
} else {
|
||||
// Ingen referanser ble dispatch’et → IKKE oppdater lastSeenTime
|
||||
log.debug { "No dispatches performed; lastSeenTime remains $lastSeenTime" }
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@ -4,7 +4,7 @@ import java.util.UUID
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
abstract class Event {
|
||||
lateinit var referenceId: UUID
|
||||
var referenceId: UUID = UUID.randomUUID()
|
||||
protected set
|
||||
var eventId: UUID = UUID.randomUUID()
|
||||
private set
|
||||
@ -39,10 +39,9 @@ inline fun <reified T> Event.requireAs(): T {
|
||||
return this as? T ?: throw IllegalArgumentException("Expected ${T::class.java.name}, got ${this::class.java.name}")
|
||||
}
|
||||
|
||||
abstract class DeleteEvent(
|
||||
open val deletedEventId: UUID
|
||||
) : Event()
|
||||
|
||||
abstract class SignalEvent(): Event()
|
||||
abstract class DeleteEvent: Event() {
|
||||
open lateinit var deletedEventId: UUID
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
@ -1,11 +1,11 @@
|
||||
package no.iktdev.eventi.models
|
||||
|
||||
import no.iktdev.eventi.MyTime
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
class Metadata {
|
||||
val created: Instant = MyTime.utcNow()
|
||||
val created: LocalDateTime = MyTime.UtcNow()
|
||||
var derivedFromId: Set<UUID>? = null
|
||||
private set
|
||||
fun derivedFromEventId(vararg id: UUID) = apply {
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
package no.iktdev.eventi.models
|
||||
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
package no.iktdev.eventi.models.store
|
||||
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
data class PersistedEvent(
|
||||
@ -9,5 +9,5 @@ data class PersistedEvent(
|
||||
val eventId: UUID,
|
||||
val event: String,
|
||||
val data: String,
|
||||
val persistedAt: Instant
|
||||
val persistedAt: LocalDateTime
|
||||
)
|
||||
@ -1,6 +1,6 @@
|
||||
package no.iktdev.eventi.models.store
|
||||
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
data class PersistedTask(
|
||||
@ -13,14 +13,13 @@ data class PersistedTask(
|
||||
val claimed: Boolean,
|
||||
val claimedBy: String? = null,
|
||||
val consumed: Boolean,
|
||||
val lastCheckIn: Instant? = null,
|
||||
val persistedAt: Instant
|
||||
val lastCheckIn: LocalDateTime? = null,
|
||||
val persistedAt: LocalDateTime
|
||||
) {}
|
||||
|
||||
enum class TaskStatus {
|
||||
Pending,
|
||||
InProgress,
|
||||
Completed,
|
||||
Failed,
|
||||
Cancelled
|
||||
Failed
|
||||
}
|
||||
@ -2,11 +2,11 @@ package no.iktdev.eventi.stores
|
||||
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.store.PersistedEvent
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
interface EventStore {
|
||||
fun getPersistedEventsAfter(timestamp: Instant): List<PersistedEvent>
|
||||
fun getPersistedEventsAfter(timestamp: LocalDateTime): List<PersistedEvent>
|
||||
fun getPersistedEventsFor(referenceId: UUID): List<PersistedEvent>
|
||||
fun persist(event: Event)
|
||||
}
|
||||
|
||||
@ -4,12 +4,12 @@ import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.currentCoroutineContext
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Task
|
||||
import no.iktdev.eventi.models.store.TaskStatus
|
||||
import org.jetbrains.annotations.VisibleForTesting
|
||||
import java.util.UUID
|
||||
import kotlin.coroutines.cancellation.CancellationException
|
||||
@ -68,16 +68,12 @@ abstract class TaskListener(val taskType: TaskType = TaskType.CPU_INTENSIVE): Ta
|
||||
currentJob = getDispatcherForTask(task).launch {
|
||||
try {
|
||||
val result = onTask(task)
|
||||
reporter.markConsumed(task.taskId)
|
||||
onComplete(task, result)
|
||||
} catch (e: CancellationException) {
|
||||
// Dette er en ekte kansellering
|
||||
onCancelled(task)
|
||||
throw e // viktig: ikke svelg cancellation
|
||||
|
||||
onCancelled()
|
||||
} catch (e: Exception) {
|
||||
// Dette er en faktisk feil
|
||||
onError(task, e)
|
||||
|
||||
} finally {
|
||||
heartbeatRunner?.cancel()
|
||||
currentJob?.cancel()
|
||||
@ -90,29 +86,24 @@ abstract class TaskListener(val taskType: TaskType = TaskType.CPU_INTENSIVE): Ta
|
||||
return true
|
||||
}
|
||||
|
||||
abstract fun createIncompleteStateTaskEvent(task: Task, status: TaskStatus, exception: Exception? = null): Event
|
||||
|
||||
override fun onError(task: Task, exception: Exception) {
|
||||
reporter?.log(task.taskId, "Error processing task: ${exception.message}")
|
||||
exception.printStackTrace()
|
||||
reporter?.markFailed(task.referenceId, task.taskId)
|
||||
reporter!!.publishEvent(createIncompleteStateTaskEvent(task, TaskStatus.Failed, exception))
|
||||
reporter?.markConsumed(task.taskId)
|
||||
}
|
||||
|
||||
override fun onComplete(task: Task, result: Event?) {
|
||||
reporter!!.markCompleted(task.taskId)
|
||||
reporter!!.markConsumed(task.taskId)
|
||||
reporter!!.log(task.taskId, "Task completed successfully.")
|
||||
result?.let {
|
||||
reporter!!.publishEvent(result)
|
||||
}
|
||||
}
|
||||
|
||||
override fun onCancelled(task: Task) {
|
||||
reporter!!.markCancelled(task.referenceId, task.taskId)
|
||||
override fun onCancelled() {
|
||||
currentJob?.cancel()
|
||||
heartbeatRunner?.cancel()
|
||||
currentTask = null
|
||||
reporter!!.publishEvent(createIncompleteStateTaskEvent(task, TaskStatus.Cancelled))
|
||||
}
|
||||
}
|
||||
|
||||
@ -129,15 +120,13 @@ interface TaskListenerImplementation {
|
||||
suspend fun onTask(task: Task): Event?
|
||||
fun onComplete(task: Task, result: Event?)
|
||||
fun onError(task: Task, exception: Exception)
|
||||
fun onCancelled(task: Task)
|
||||
fun onCancelled()
|
||||
}
|
||||
|
||||
interface TaskReporter {
|
||||
fun markClaimed(taskId: UUID, workerId: String)
|
||||
fun updateLastSeen(taskId: UUID)
|
||||
fun markCompleted(taskId: UUID)
|
||||
fun markFailed(referenceId: UUID, taskId: UUID)
|
||||
fun markCancelled(referenceId: UUID, taskId: UUID)
|
||||
fun markConsumed(taskId: UUID)
|
||||
fun updateProgress(taskId: UUID, progress: Int)
|
||||
fun log(taskId: UUID, message: String)
|
||||
fun publishEvent(event: Event)
|
||||
|
||||
@ -8,7 +8,7 @@ import no.iktdev.eventi.events.EventListenerRegistry
|
||||
import no.iktdev.eventi.events.EventTypeRegistry
|
||||
import no.iktdev.eventi.models.DeleteEvent
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.SignalEvent
|
||||
import no.iktdev.eventi.models.Metadata
|
||||
import no.iktdev.eventi.testUtil.wipe
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
@ -18,53 +18,43 @@ import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
@DisplayName(
|
||||
"""
|
||||
EventDispatcher
|
||||
Når hendelser dispatches til lyttere
|
||||
Hvis hendelsene inneholder avledede, slettede eller nye events
|
||||
Så skal dispatcheren håndtere filtrering, replays og historikk korrekt
|
||||
"""
|
||||
)
|
||||
class EventDispatcherTest : TestBase() {
|
||||
|
||||
class EventDispatcherTest: TestBase() {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
|
||||
class DerivedEvent : Event()
|
||||
class TriggerEvent : Event()
|
||||
class OtherEvent : Event()
|
||||
class DummyEvent : Event() {
|
||||
class DerivedEvent(): Event()
|
||||
class TriggerEvent(): Event() {
|
||||
}
|
||||
class OtherEvent(): Event()
|
||||
class DummyEvent(): Event() {
|
||||
fun putMetadata(metadata: Metadata) {
|
||||
this.metadata = metadata
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@BeforeEach
|
||||
fun setup() {
|
||||
EventTypeRegistry.wipe()
|
||||
EventListenerRegistry.wipe()
|
||||
// Verifiser at det er tomt
|
||||
|
||||
EventTypeRegistry.register(
|
||||
listOf(
|
||||
DerivedEvent::class.java,
|
||||
TriggerEvent::class.java,
|
||||
OtherEvent::class.java,
|
||||
DummyEvent::class.java
|
||||
)
|
||||
)
|
||||
EventTypeRegistry.register(listOf(
|
||||
DerivedEvent::class.java,
|
||||
TriggerEvent::class.java,
|
||||
OtherEvent::class.java,
|
||||
DummyEvent::class.java
|
||||
))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en TriggerEvent dispatches
|
||||
Hvis en lytter produserer én DerivedEvent
|
||||
Så skal kun én ny event produseres og prosessen stoppe
|
||||
"""
|
||||
)
|
||||
fun shouldProduceOneEventAndStop() {
|
||||
ProducingListener()
|
||||
|
||||
val trigger = TriggerEvent().newReferenceId()
|
||||
@Test
|
||||
fun `should produce one event and stop`() {
|
||||
val listener = ProducingListener()
|
||||
|
||||
val trigger = TriggerEvent()
|
||||
dispatcher.dispatch(trigger.referenceId, listOf(trigger))
|
||||
|
||||
val produced = eventStore.all().firstOrNull()
|
||||
@ -72,85 +62,52 @@ class EventDispatcherTest : TestBase() {
|
||||
|
||||
val event = produced!!.toEvent()
|
||||
assertThat(event!!.metadata.derivedFromId).hasSize(1)
|
||||
assertThat(event.metadata.derivedFromId).contains(trigger.eventId)
|
||||
assertThat(event!!.metadata.derivedFromId).contains(trigger.eventId)
|
||||
assertTrue(event is DerivedEvent)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en event allerede har avledet en DerivedEvent
|
||||
Hvis dispatcheren replays historikken
|
||||
Så skal ikke DerivedEvent produseres på nytt
|
||||
"""
|
||||
)
|
||||
fun shouldSkipAlreadyDerivedEvents() {
|
||||
ProducingListener()
|
||||
fun `should skip already derived events`() {
|
||||
val listener = ProducingListener()
|
||||
|
||||
val trigger = TriggerEvent().newReferenceId()
|
||||
val derived = DerivedEvent().derivedOf(trigger).toPersisted(1L, MyTime.utcNow())
|
||||
val trigger = TriggerEvent()
|
||||
val derived = DerivedEvent().derivedOf(trigger).toPersisted(1L, MyTime.UtcNow())
|
||||
|
||||
eventStore.persist(derived!!.toEvent()!!) // simulate prior production
|
||||
|
||||
dispatcher.dispatch(trigger.referenceId, listOf(trigger, derived.toEvent()!!))
|
||||
dispatcher.dispatch(trigger.referenceId, listOf(trigger, derived!!.toEvent()!!))
|
||||
|
||||
assertEquals(1, eventStore.all().size)
|
||||
assertEquals(1, eventStore.all().size) // no new event produced
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når flere events dispatches
|
||||
Hvis en lytter mottar en event
|
||||
Så skal hele historikken leveres i context
|
||||
"""
|
||||
)
|
||||
fun shouldPassFullContextToListener() {
|
||||
fun `should pass full context to listener`() {
|
||||
val listener = ContextCapturingListener()
|
||||
|
||||
val e1 = TriggerEvent().newReferenceId()
|
||||
val e2 = OtherEvent().newReferenceId()
|
||||
val e1 = TriggerEvent()
|
||||
val e2 = OtherEvent()
|
||||
dispatcher.dispatch(e1.referenceId, listOf(e1, e2))
|
||||
|
||||
assertEquals(2, listener.context.size)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en replay skjer
|
||||
Hvis en event allerede har produsert en DerivedEvent
|
||||
Så skal ikke DerivedEvent produseres på nytt
|
||||
"""
|
||||
)
|
||||
fun shouldBehaveDeterministicallyAcrossReplays() {
|
||||
val referenceId = UUID.randomUUID()
|
||||
fun `should behave deterministically across replays`() {
|
||||
val listener = ProducingListener()
|
||||
|
||||
ProducingListener()
|
||||
|
||||
val trigger = TriggerEvent().usingReferenceId(referenceId)
|
||||
val trigger = TriggerEvent()
|
||||
dispatcher.dispatch(trigger.referenceId, listOf(trigger))
|
||||
val replayContext = listOf(trigger) + eventStore.all().mapNotNull { it.toEvent() }
|
||||
|
||||
dispatcher.dispatch(trigger.referenceId, replayContext)
|
||||
|
||||
assertEquals(1, eventStore.all().size)
|
||||
assertEquals(1, eventStore.all().size) // no duplicate
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en DeleteEvent peker på en tidligere event
|
||||
Hvis dispatcheren filtrerer kandidater
|
||||
Så skal slettede events ikke leveres som kandidater
|
||||
"""
|
||||
)
|
||||
fun shouldNotDeliverDeletedEventsAsCandidates() {
|
||||
val referenceId = UUID.randomUUID()
|
||||
|
||||
fun `should not deliver deleted events as candidates`() {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
val received = mutableListOf<Event>()
|
||||
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
received += event
|
||||
@ -158,10 +115,12 @@ class EventDispatcherTest : TestBase() {
|
||||
}
|
||||
}
|
||||
// Original hendelse
|
||||
val original = TriggerEvent().usingReferenceId(referenceId)
|
||||
val original = TriggerEvent()
|
||||
|
||||
// Slettehendelse som peker på original
|
||||
val deleted = object : DeleteEvent(original.eventId) {}.apply { newReferenceId() }
|
||||
val deleted = object : DeleteEvent() {
|
||||
override var deletedEventId = original.eventId
|
||||
}
|
||||
|
||||
// Dispatch med begge hendelser
|
||||
dispatcher.dispatch(original.referenceId, listOf(original, deleted))
|
||||
@ -178,42 +137,29 @@ class EventDispatcherTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en DeleteEvent dispatches alene
|
||||
Hvis en lytter reagerer på DeleteEvent
|
||||
Så skal DeleteEvent leveres som kandidat
|
||||
"""
|
||||
)
|
||||
fun shouldDeliverDeleteEventToListenersThatReactToIt() {
|
||||
fun `should deliver DeleteEvent to listeners that react to it`() {
|
||||
val received = mutableListOf<Event>()
|
||||
val referenceId = UUID.randomUUID()
|
||||
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
val listener = object : EventListener() {
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
if (event is DeleteEvent) received += event
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
val deleted = object : DeleteEvent(UUID.randomUUID()) {}.apply { usingReferenceId(referenceId) }
|
||||
val deleted = object : DeleteEvent() {
|
||||
override var deletedEventId = UUID.randomUUID()
|
||||
}
|
||||
dispatcher.dispatch(deleted.referenceId, listOf(deleted))
|
||||
|
||||
assertTrue(received.contains(deleted))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en event har avledet en ny event
|
||||
Hvis dispatcheren replays historikken
|
||||
Så skal ikke original-eventen leveres som kandidat igjen
|
||||
"""
|
||||
)
|
||||
fun shouldNotRedeliverEventsThatHaveProducedDerivedEvents() {
|
||||
ProducingListener()
|
||||
@DisplayName("Replay skal ikke levere en event som allerede har avledet en ny")
|
||||
fun `should not re-deliver events that have produced derived events`() {
|
||||
val listener = ProducingListener()
|
||||
|
||||
val trigger = TriggerEvent().newReferenceId()
|
||||
val trigger = TriggerEvent()
|
||||
// Første dispatch: trigger produserer en DerivedEvent
|
||||
dispatcher.dispatch(trigger.referenceId, listOf(trigger))
|
||||
|
||||
@ -232,157 +178,20 @@ class EventDispatcherTest : TestBase() {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en DeleteEvent slettet en tidligere event
|
||||
Hvis dispatcheren bygger historikk
|
||||
Så skal slettede events ikke være med i history
|
||||
"""
|
||||
)
|
||||
fun historyShouldExcludeDeletedEvents() {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
|
||||
val original = TriggerEvent().newReferenceId()
|
||||
val deleted = object : DeleteEvent(original.eventId) {}.apply { usingReferenceId(original.referenceId) }
|
||||
|
||||
var receivedHistory: List<Event> = emptyList()
|
||||
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
receivedHistory = history
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
dispatcher.dispatch(original.referenceId, listOf(original, deleted))
|
||||
|
||||
assertFalse(receivedHistory.contains(original))
|
||||
assertFalse(receivedHistory.contains(deleted))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en DeleteEvent slettet en event
|
||||
Hvis andre events fortsatt er gyldige
|
||||
Så skal history kun inneholde de ikke-slettede events
|
||||
"""
|
||||
)
|
||||
fun historyShouldKeepNonDeletedEvents() {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
val referenceId = UUID.randomUUID()
|
||||
val e1 = TriggerEvent().usingReferenceId(referenceId)
|
||||
val e2 = OtherEvent().usingReferenceId(referenceId)
|
||||
val deleted = object : DeleteEvent(e1.eventId) {}
|
||||
|
||||
var receivedHistory: List<Event> = emptyList()
|
||||
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
receivedHistory = history
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
dispatcher.dispatch(e1.referenceId, listOf(e1, e2, deleted))
|
||||
|
||||
assertTrue(receivedHistory.contains(e2))
|
||||
assertFalse(receivedHistory.contains(e1))
|
||||
assertFalse(receivedHistory.contains(deleted))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en DeleteEvent er kandidat
|
||||
Hvis historikken kun inneholder slettede events
|
||||
Så skal history være tom
|
||||
"""
|
||||
)
|
||||
fun deleteEventShouldBeDeliveredButHistoryEmpty() {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
|
||||
val original = TriggerEvent().newReferenceId()
|
||||
val deleted = object : DeleteEvent(original.eventId) {}.apply { newReferenceId() }
|
||||
|
||||
var receivedEvent: Event? = null
|
||||
var receivedHistory: List<Event> = emptyList()
|
||||
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
receivedEvent = event
|
||||
receivedHistory = history
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
dispatcher.dispatch(original.referenceId, listOf(original, deleted))
|
||||
|
||||
assertTrue(receivedEvent is DeleteEvent)
|
||||
assertTrue(receivedHistory.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName(
|
||||
"""
|
||||
Når en SignalEvent dispatches
|
||||
Hvis SignalEvent ikke skal være kandidat
|
||||
Så skal den ikke leveres til lyttere, men fortsatt være i historikken
|
||||
"""
|
||||
)
|
||||
fun shouldNotDeliverSignalEventAsCandidate() {
|
||||
// Arrange
|
||||
class TestSignalEvent : SignalEvent()
|
||||
EventTypeRegistry.register(listOf(TestSignalEvent::class.java,))
|
||||
|
||||
val received = mutableListOf<Event>()
|
||||
var finalHistory: List<Event>? = null
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
received += event
|
||||
finalHistory = history
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
val refId = UUID.randomUUID()
|
||||
val trigger = TriggerEvent().usingReferenceId(refId)
|
||||
val signal = TestSignalEvent().usingReferenceId(refId)
|
||||
|
||||
// Act
|
||||
dispatcher.dispatch(trigger.referenceId, listOf(trigger, signal))
|
||||
|
||||
// Assert
|
||||
// 1) TriggerEvent skal leveres
|
||||
assertTrue(received.any { it is TriggerEvent }) {
|
||||
"TriggerEvent skal leveres som kandidat"
|
||||
}
|
||||
|
||||
// 2) SignalEvent skal IKKE leveres
|
||||
assertFalse(received.any { it is TestSignalEvent }) {
|
||||
"SignalEvent skal ikke leveres som kandidat"
|
||||
}
|
||||
|
||||
assertNotNull(finalHistory)
|
||||
assertTrue(finalHistory!!.any { it is TestSignalEvent }) {
|
||||
"SignalEvent skal være i historikken selv om den ikke er kandidat"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// --- Test helpers ---
|
||||
|
||||
class ProducingListener : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
return if (event is TriggerEvent) DerivedEvent().derivedOf(event) else null
|
||||
}
|
||||
}
|
||||
|
||||
class ContextCapturingListener : EventListener() {
|
||||
var context: List<Event> = emptyList()
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
this.context = history
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
this.context = context
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
@ -4,25 +4,25 @@ import no.iktdev.eventi.ZDS.toPersisted
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.store.PersistedEvent
|
||||
import no.iktdev.eventi.stores.EventStore
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
class InMemoryEventStore : EventStore {
|
||||
private val persisted = mutableListOf<PersistedEvent>()
|
||||
private var nextId = 1L
|
||||
|
||||
override fun getPersistedEventsAfter(timestamp: Instant): List<PersistedEvent> =
|
||||
override fun getPersistedEventsAfter(timestamp: LocalDateTime): List<PersistedEvent> =
|
||||
persisted.filter { it.persistedAt > timestamp }
|
||||
|
||||
override fun getPersistedEventsFor(referenceId: UUID): List<PersistedEvent> =
|
||||
persisted.filter { it.referenceId == referenceId }
|
||||
|
||||
override fun persist(event: Event) {
|
||||
val persistedEvent = event.toPersisted(nextId++, MyTime.utcNow())
|
||||
val persistedEvent = event.toPersisted(nextId++, MyTime.UtcNow())
|
||||
persisted += persistedEvent!!
|
||||
}
|
||||
|
||||
fun persistAt(event: Event, persistedAt: Instant) {
|
||||
fun persistAt(event: Event, persistedAt: LocalDateTime) {
|
||||
val persistedEvent = event.toPersisted(nextId++, persistedAt)
|
||||
persisted += persistedEvent!!
|
||||
}
|
||||
|
||||
@ -6,7 +6,7 @@ import no.iktdev.eventi.models.store.PersistedTask
|
||||
import no.iktdev.eventi.models.store.TaskStatus
|
||||
import no.iktdev.eventi.stores.TaskStore
|
||||
import java.time.Duration
|
||||
import java.time.temporal.ChronoUnit
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
import kotlin.concurrent.atomics.AtomicReference
|
||||
|
||||
@ -30,13 +30,13 @@ open class InMemoryTaskStore : TaskStore {
|
||||
override fun claim(taskId: UUID, workerId: String): Boolean {
|
||||
val task = findByTaskId(taskId) ?: return false
|
||||
if (task.claimed && !isExpired(task)) return false
|
||||
update(task.copy(claimed = true, claimedBy = workerId, lastCheckIn = MyTime.utcNow()))
|
||||
update(task.copy(claimed = true, claimedBy = workerId, lastCheckIn = MyTime.UtcNow()))
|
||||
return true
|
||||
}
|
||||
|
||||
override fun heartbeat(taskId: UUID) {
|
||||
val task = findByTaskId(taskId) ?: return
|
||||
update(task.copy(lastCheckIn = MyTime.utcNow()))
|
||||
update(task.copy(lastCheckIn = MyTime.UtcNow()))
|
||||
}
|
||||
|
||||
override fun markConsumed(taskId: UUID, status: TaskStatus) {
|
||||
@ -45,7 +45,7 @@ open class InMemoryTaskStore : TaskStore {
|
||||
}
|
||||
|
||||
override fun releaseExpiredTasks(timeout: Duration) {
|
||||
val now = MyTime.utcNow()
|
||||
val now = MyTime.UtcNow()
|
||||
tasks.filter {
|
||||
it.claimed && !it.consumed && it.lastCheckIn?.isBefore(now.minus(timeout)) == true
|
||||
}.forEach {
|
||||
@ -60,8 +60,8 @@ open class InMemoryTaskStore : TaskStore {
|
||||
}
|
||||
|
||||
private fun isExpired(task: PersistedTask): Boolean {
|
||||
val now = MyTime.utcNow()
|
||||
return task.lastCheckIn?.isBefore(now.minus(15, ChronoUnit.MINUTES)) == true
|
||||
val now = MyTime.UtcNow()
|
||||
return task.lastCheckIn?.isBefore(now.minusMinutes(15)) == true
|
||||
}
|
||||
|
||||
private fun serialize(data: Any?): String = data?.toString() ?: "{}"
|
||||
|
||||
@ -12,52 +12,40 @@ import org.junit.jupiter.api.Assertions.assertNull
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.LocalDateTime
|
||||
|
||||
@DisplayName("""
|
||||
ZDS – Serialization/Deserialization System
|
||||
Når Event- og Task-objekter persisteres og gjenopprettes
|
||||
Hvis type-registrene er korrekt konfigurert
|
||||
Så skal ZDS kunne serialisere og deserialisere objektene uten tap av data
|
||||
""")
|
||||
class ZDSTest {
|
||||
|
||||
@BeforeEach
|
||||
fun setup() {
|
||||
EventTypeRegistry.wipe()
|
||||
TaskTypeRegistry.wipe()
|
||||
|
||||
// Verifiser at det er tomt
|
||||
assertNull(EventTypeRegistry.resolve("SomeEvent"))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når et Event-objekt persisteres via ZDS
|
||||
Hvis typen er registrert i EventTypeRegistry
|
||||
Så skal det kunne gjenopprettes som riktig Event-type med samme data
|
||||
""")
|
||||
@DisplayName("Test ZDS with Event object")
|
||||
fun scenario1() {
|
||||
EventTypeRegistry.register(EchoEvent::class.java)
|
||||
|
||||
val echo = EchoEvent("hello").newReferenceId()
|
||||
val echo = EchoEvent("hello")
|
||||
val persisted = echo.toPersisted(id = 1L)
|
||||
|
||||
val restored = persisted!!.toEvent()
|
||||
assert(restored is EchoEvent)
|
||||
assert((restored as EchoEvent).data == "hello")
|
||||
|
||||
}
|
||||
|
||||
data class TestTask(
|
||||
val data: String?
|
||||
) : Task()
|
||||
): Task()
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når et Task-objekt persisteres via ZDS
|
||||
Hvis typen er registrert i TaskTypeRegistry
|
||||
Så skal det kunne gjenopprettes som riktig Task-type med metadata intakt
|
||||
""")
|
||||
@DisplayName("Test ZDS with Task object")
|
||||
fun scenario2() {
|
||||
|
||||
TaskTypeRegistry.register(TestTask::class.java)
|
||||
|
||||
val task = TestTask("Potato")
|
||||
@ -70,5 +58,7 @@ class ZDSTest {
|
||||
assert((restored as TestTask).data == "Potato")
|
||||
assert(restored.metadata.created == task.metadata.created)
|
||||
assert(restored.metadata.derivedFromId == task.metadata.derivedFromId)
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -6,34 +6,35 @@ import no.iktdev.eventi.testUtil.wipe
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.Assertions.*
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
@DisplayName("""
|
||||
EventListenerRegistry
|
||||
Når lyttere registreres med og uten @ListenerOrder
|
||||
Hvis registry sorterer dem etter annotasjonen
|
||||
Så skal rekkefølgen være deterministisk og korrekt
|
||||
""")
|
||||
class EventListenerRegistryTest {
|
||||
|
||||
@ListenerOrder(1)
|
||||
class MockTest1 : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? = null
|
||||
class MockTest1() : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
@ListenerOrder(2)
|
||||
class MockTest2 : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? = null
|
||||
class MockTest2() : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
@ListenerOrder(3)
|
||||
class MockTest3 : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? = null
|
||||
class MockTest3() : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
class MockTestRandom : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? = null
|
||||
class MockTestRandom() : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
@BeforeEach
|
||||
@ -42,24 +43,19 @@ class EventListenerRegistryTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når flere lyttere registreres i vilkårlig rekkefølge
|
||||
Hvis noen har @ListenerOrder og andre ikke
|
||||
Så skal registry returnere dem sortert etter order, og usorterte sist
|
||||
""")
|
||||
fun validateOrder() {
|
||||
MockTestRandom()
|
||||
MockTest1()
|
||||
MockTest2()
|
||||
MockTest3()
|
||||
|
||||
val listeners = EventListenerRegistry.getListeners()
|
||||
|
||||
// Assert
|
||||
assertThat(listeners.map { it::class.simpleName }).containsExactly(
|
||||
MockTest1::class.simpleName, // @ListenerOrder(1)
|
||||
MockTest2::class.simpleName, // @ListenerOrder(2)
|
||||
MockTest3::class.simpleName, // @ListenerOrder(3)
|
||||
MockTest1::class.simpleName, // @ListenerOrder(1)
|
||||
MockTest2::class.simpleName, // @ListenerOrder(2)
|
||||
MockTest3::class.simpleName, // @ListenerOrder(3)
|
||||
MockTestRandom::class.simpleName // no annotation → goes last
|
||||
)
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,107 +1,81 @@
|
||||
package no.iktdev.eventi.events
|
||||
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineDispatcher
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.test.StandardTestDispatcher
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.coroutines.withContext
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import no.iktdev.eventi.EventDispatcherTest
|
||||
import no.iktdev.eventi.EventDispatcherTest.DerivedEvent
|
||||
import no.iktdev.eventi.EventDispatcherTest.OtherEvent
|
||||
import no.iktdev.eventi.EventDispatcherTest.TriggerEvent
|
||||
import no.iktdev.eventi.MyTime
|
||||
import no.iktdev.eventi.TestBase
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.testUtil.TestSequenceDispatchQueue
|
||||
import no.iktdev.eventi.testUtil.wipe
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertFalse
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.Duration
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
@DisplayName("""
|
||||
EventPollerImplementation
|
||||
Når polleren leser nye events fra EventStore og samarbeider med SequenceDispatchQueue
|
||||
Hvis nye events ankommer, køen er travel, eller duplikater dukker opp
|
||||
Så skal polleren dispatch'e riktig, oppdatere lastSeenTime og unngå duplikater
|
||||
""")
|
||||
class EventPollerImplementationTest : TestBase() {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
val queue = SequenceDispatchQueue(maxConcurrency = 8)
|
||||
|
||||
private val dispatcher = EventDispatcher(eventStore)
|
||||
val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
|
||||
|
||||
@BeforeEach
|
||||
fun setup() {
|
||||
EventTypeRegistry.wipe()
|
||||
EventListenerRegistry.wipe()
|
||||
eventStore.clear()
|
||||
// Verifiser at det er tomt
|
||||
|
||||
EventTypeRegistry.register(
|
||||
listOf(
|
||||
DerivedEvent::class.java,
|
||||
TriggerEvent::class.java,
|
||||
OtherEvent::class.java
|
||||
)
|
||||
)
|
||||
EventTypeRegistry.register(listOf(
|
||||
DerivedEvent::class.java,
|
||||
TriggerEvent::class.java,
|
||||
OtherEvent::class.java
|
||||
))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren finner nye referenceId-er med events
|
||||
Hvis pollOnce kjøres
|
||||
Så skal alle referenceId-er dispatch'es og lastSeenTime oppdateres
|
||||
""")
|
||||
fun pollOnceDispatchesAllNewReferenceIdsAndUpdatesLastSeenTime() = runTest {
|
||||
val testDispatcher = StandardTestDispatcher(testScheduler)
|
||||
val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
|
||||
val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
|
||||
|
||||
fun `pollOnce should dispatch all new referenceIds and update lastSeenTime`() = runTest {
|
||||
val dispatched = ConcurrentHashMap.newKeySet<UUID>()
|
||||
val completionMap = mutableMapOf<UUID, CompletableDeferred<Unit>>()
|
||||
|
||||
EventListenerRegistry.registerListener(
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
dispatched += event.referenceId
|
||||
completionMap[event.referenceId]?.complete(Unit)
|
||||
return null
|
||||
}
|
||||
EventListenerRegistry.registerListener(object : EventListener() {
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
dispatched += event.referenceId
|
||||
completionMap[event.referenceId]?.complete(Unit)
|
||||
return null
|
||||
}
|
||||
)
|
||||
})
|
||||
|
||||
val referenceIds = (1..10).map { UUID.randomUUID() }
|
||||
|
||||
referenceIds.forEach { refId ->
|
||||
val e = TriggerEvent().usingReferenceId(refId)
|
||||
eventStore.persist(e)
|
||||
val e = EventDispatcherTest.TriggerEvent().usingReferenceId(refId)
|
||||
eventStore.persist(e) // persistedAt settes automatisk her
|
||||
completionMap[refId] = CompletableDeferred()
|
||||
}
|
||||
|
||||
poller.pollOnce()
|
||||
|
||||
completionMap.values.awaitAll()
|
||||
|
||||
assertEquals(referenceIds.toSet(), dispatched)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren ikke finner nye events
|
||||
Hvis pollOnce kjøres flere ganger
|
||||
Så skal backoff øke, og resettes når nye events ankommer
|
||||
""")
|
||||
fun pollOnceIncreasesBackoffWhenNoEventsAndResetsWhenEventsArrive() = runTest {
|
||||
val testDispatcher = StandardTestDispatcher(testScheduler)
|
||||
val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
|
||||
|
||||
fun `pollOnce should increase backoff when no events and reset when events arrive`() = runTest {
|
||||
val testPoller = object : EventPollerImplementation(eventStore, queue, dispatcher) {
|
||||
fun currentBackoff(): Duration = backoff
|
||||
}
|
||||
@ -124,28 +98,20 @@ class EventPollerImplementationTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når flere events med samme referenceId ligger i EventStore
|
||||
Hvis pollOnce kjøres
|
||||
Så skal polleren gruppere og dispatch'e alle tre i én batch
|
||||
""")
|
||||
fun pollOnceGroupsAndDispatchesExactlyThreeEventsForOneReferenceId() = runTest {
|
||||
val testDispatcher = StandardTestDispatcher(testScheduler)
|
||||
val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
|
||||
val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
|
||||
|
||||
fun `pollOnce should group and dispatch exactly 3 events for one referenceId`() = runTest {
|
||||
val refId = UUID.randomUUID()
|
||||
val received = mutableListOf<Event>()
|
||||
val done = CompletableDeferred<Unit>()
|
||||
|
||||
// Wipe alt før test
|
||||
EventTypeRegistry.wipe()
|
||||
EventListenerRegistry.wipe()
|
||||
eventStore.clear()
|
||||
eventStore.clear() // sørg for at InMemoryEventStore støtter dette
|
||||
|
||||
EventTypeRegistry.register(listOf(TriggerEvent::class.java))
|
||||
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
received += event
|
||||
if (received.size == 3) done.complete(Unit)
|
||||
return null
|
||||
@ -157,32 +123,27 @@ class EventPollerImplementationTest : TestBase() {
|
||||
}
|
||||
|
||||
poller.pollOnce()
|
||||
|
||||
done.await()
|
||||
|
||||
assertEquals(3, received.size)
|
||||
assertTrue(received.all { it.referenceId == refId })
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren har en lastSeenTime i fremtiden
|
||||
Hvis events ankommer med eldre timestamp
|
||||
Så skal polleren ignorere dem
|
||||
""")
|
||||
fun pollOnceIgnoresEventsBeforeLastSeenTime() = runTest {
|
||||
val testDispatcher = StandardTestDispatcher(testScheduler)
|
||||
val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
|
||||
|
||||
val testPoller = object : EventPollerImplementation(eventStore, queue, dispatcher) {
|
||||
init {
|
||||
lastSeenTime = MyTime.utcNow().plusSeconds(1)
|
||||
}
|
||||
}
|
||||
|
||||
fun `pollOnce should ignore events before lastSeenTime`() = runTest {
|
||||
val refId = UUID.randomUUID()
|
||||
val ignored = TriggerEvent().usingReferenceId(refId)
|
||||
|
||||
val testPoller = object : EventPollerImplementation(eventStore, queue, dispatcher) {
|
||||
init {
|
||||
lastSeenTime = MyTime.UtcNow().plusSeconds(1)
|
||||
}
|
||||
}
|
||||
|
||||
eventStore.persist(ignored)
|
||||
|
||||
testPoller.pollOnce()
|
||||
|
||||
assertFalse(queue.isProcessing(refId))
|
||||
@ -190,53 +151,61 @@ class EventPollerImplementationTest : TestBase() {
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en duplikat-event injiseres manuelt i EventStore
|
||||
Hvis polleren kjører igjen
|
||||
Så skal begge events prosesseres, men uten å produsere duplikate derived events
|
||||
""")
|
||||
fun pollerHandlesManuallyInjectedDuplicateEvent() = runTest {
|
||||
val testDispatcher = StandardTestDispatcher(testScheduler)
|
||||
val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
|
||||
val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
|
||||
|
||||
fun `poller handles manually injected duplicate event`() = runTest {
|
||||
EventTypeRegistry.register(listOf(MarcoEvent::class.java, EchoEvent::class.java))
|
||||
|
||||
val channel = Channel<Event>(Channel.UNLIMITED)
|
||||
val handled = mutableListOf<Event>()
|
||||
|
||||
|
||||
// Setup
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
if (event !is EchoEvent) return null
|
||||
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
if (event !is EchoEvent)
|
||||
return null
|
||||
handled += event
|
||||
channel.trySend(event)
|
||||
return MarcoEvent(true).derivedOf(event)
|
||||
}
|
||||
}
|
||||
|
||||
val original = EchoEvent("Hello").newReferenceId()
|
||||
val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {
|
||||
}
|
||||
|
||||
// Original event
|
||||
val original = EchoEvent(data = "Hello")
|
||||
eventStore.persist(original)
|
||||
|
||||
// Act
|
||||
poller.pollOnce()
|
||||
|
||||
withContext(testDispatcher) {
|
||||
withTimeout(60_000) {
|
||||
channel.receive()
|
||||
withContext(Dispatchers.Default.limitedParallelism(1)) {
|
||||
withTimeout(Duration.ofMinutes(1).toMillis()) {
|
||||
repeat(1) { channel.receive() }
|
||||
}
|
||||
}
|
||||
|
||||
// Manual replay with new eventId, same referenceId
|
||||
val duplicateEvent = EchoEvent("Test me").usingReferenceId(original.referenceId)
|
||||
|
||||
eventStore.persist(duplicateEvent)
|
||||
|
||||
// Act
|
||||
poller.pollOnce()
|
||||
|
||||
withContext(testDispatcher) {
|
||||
withTimeout(60_000) {
|
||||
channel.receive()
|
||||
withContext(Dispatchers.Default.limitedParallelism(1)) {
|
||||
withTimeout(Duration.ofMinutes(1).toMillis()) {
|
||||
repeat(1) { channel.receive() }
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
// Assert
|
||||
assertEquals(2, handled.size)
|
||||
assertTrue(handled.any { it.eventId == original.eventId })
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
@ -1,18 +1,19 @@
|
||||
@file:OptIn(ExperimentalCoroutinesApi::class)
|
||||
|
||||
package no.iktdev.eventi.events
|
||||
|
||||
import io.mockk.every
|
||||
import io.mockk.mockk
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.test.*
|
||||
import no.iktdev.eventi.InMemoryEventStore
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
import kotlinx.coroutines.*
|
||||
import no.iktdev.eventi.ZDS.toPersisted
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Metadata
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import java.time.Instant
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
|
||||
@ -58,12 +59,13 @@ class TestEvent : Event() {
|
||||
}
|
||||
|
||||
|
||||
@DisplayName("""
|
||||
EventPollerImplementation – simulert kø og dispatch
|
||||
Når polleren leser events fra EventStore og samarbeider med SequenceDispatchQueue
|
||||
Hvis køen er ledig, travel, eller events ankommer i ulike tidsrekkefølger
|
||||
Så skal polleren oppdatere lastSeenTime, unngå duplikater og prosessere riktig
|
||||
""")
|
||||
class FakeClock(var now: LocalDateTime) {
|
||||
fun advanceSeconds(sec: Long) {
|
||||
now = now.plusSeconds(sec)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class RunSimulationTestTest {
|
||||
|
||||
private lateinit var store: InMemoryEventStore
|
||||
@ -86,67 +88,57 @@ class RunSimulationTestTest {
|
||||
}
|
||||
}
|
||||
|
||||
private fun persistEvent(ref: UUID) {
|
||||
private fun persistEvent(ref: UUID, time: LocalDateTime) {
|
||||
val e = TestEvent().withReference(ref)
|
||||
store.persist(e.setMetadata(Metadata()))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren finner nye events
|
||||
Hvis dispatch skjer normalt
|
||||
Så skal lastSeenTime oppdateres og dispatcheren få én dispatch
|
||||
""")
|
||||
fun pollerUpdatesLastSeenTimeWhenDispatchHappens() = runTest(testDispatcher) {
|
||||
fun `poller updates lastSeenTime when dispatch happens`() = runTest(testDispatcher) {
|
||||
val ref = UUID.randomUUID()
|
||||
val t = LocalDateTime.of(2026, 1, 22, 12, 0, 0)
|
||||
|
||||
persistEvent(ref)
|
||||
persistEvent(ref, t)
|
||||
|
||||
poller.pollOnce()
|
||||
advanceUntilIdle()
|
||||
|
||||
assertThat(poller.lastSeenTime).isGreaterThan(Instant.EPOCH)
|
||||
assertThat(poller.lastSeenTime).isAfter(LocalDateTime.of(1970,1,1,0,0))
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
}
|
||||
|
||||
|
||||
class AlwaysBusyDispatchQueue : SequenceDispatchQueue(8, CoroutineScope(Dispatchers.Default)) {
|
||||
override fun isProcessing(referenceId: UUID): Boolean = true
|
||||
override fun dispatch(referenceId: UUID, events: List<Event>, dispatcher: EventDispatcher) = null
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når køen er travel og ikke kan dispatch'e
|
||||
Hvis polleren likevel ser nye events
|
||||
Så skal lastSeenTime fortsatt oppdateres (livelock-fix)
|
||||
""")
|
||||
fun pollerUpdatesLastSeenTimeEvenWhenQueueBusy() = runTest {
|
||||
fun `poller does NOT update lastSeenTime when queue is busy`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
val t = Instant.parse("2026-01-22T12:00:00Z")
|
||||
val t = LocalDateTime.of(2026,1,22,12,0,0)
|
||||
|
||||
store.persistAt(TestEvent().withReference(ref), t)
|
||||
|
||||
val busyQueue = AlwaysBusyDispatchQueue()
|
||||
|
||||
val poller = object : EventPollerImplementation(store, busyQueue, dispatcher) {}
|
||||
|
||||
poller.pollOnce()
|
||||
advanceUntilIdle()
|
||||
|
||||
// Etter livelock-fixen skal lastSeenTime være *etter* eventet
|
||||
assertThat(poller.lastSeenTime)
|
||||
.isGreaterThan(t)
|
||||
.isEqualTo(LocalDateTime.of(1970,1,1,0,0))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren kjører flere ganger uten nye events
|
||||
Hvis første poll allerede dispatch'et eventet
|
||||
Så skal polleren ikke dispatch'e samme event to ganger
|
||||
""")
|
||||
fun pollerDoesNotDoubleDispatch() = runTest(testDispatcher) {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
persistEvent(ref)
|
||||
|
||||
@Test
|
||||
fun `poller does not double-dispatch`() = runTest(testDispatcher) {
|
||||
val ref = UUID.randomUUID()
|
||||
val t = LocalDateTime.of(2026, 1, 22, 12, 0, 0)
|
||||
|
||||
persistEvent(ref, t)
|
||||
|
||||
poller.pollOnce()
|
||||
advanceUntilIdle()
|
||||
@ -158,17 +150,13 @@ class RunSimulationTestTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når flere referenceId-er har nye events
|
||||
Hvis polleren kjører én runde
|
||||
Så skal begge referenceId-er dispatch'es
|
||||
""")
|
||||
fun pollerHandlesMultipleReferenceIds() = runTest(testDispatcher) {
|
||||
fun `poller handles multiple referenceIds`() = runTest(testDispatcher) {
|
||||
val refA = UUID.randomUUID()
|
||||
val refB = UUID.randomUUID()
|
||||
val t = LocalDateTime.of(2026, 1, 22, 12, 0, 0)
|
||||
|
||||
persistEvent(refA)
|
||||
persistEvent(refB)
|
||||
persistEvent(refA, t)
|
||||
persistEvent(refB, t.plusSeconds(1))
|
||||
|
||||
poller.pollOnce()
|
||||
advanceUntilIdle()
|
||||
@ -177,17 +165,13 @@ class RunSimulationTestTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når to events har identisk timestamp
|
||||
Hvis polleren leser dem i samme poll
|
||||
Så skal begge referenceId-er dispatch'es
|
||||
""")
|
||||
fun pollerHandlesIdenticalTimestamps() = runTest(testDispatcher) {
|
||||
fun `poller handles identical timestamps`() = runTest(testDispatcher) {
|
||||
val refA = UUID.randomUUID()
|
||||
val refB = UUID.randomUUID()
|
||||
val t = LocalDateTime.of(2026, 1, 22, 12, 0, 0)
|
||||
|
||||
persistEvent(refA)
|
||||
persistEvent(refB)
|
||||
persistEvent(refA, t)
|
||||
persistEvent(refB, t)
|
||||
|
||||
poller.pollOnce()
|
||||
advanceUntilIdle()
|
||||
@ -196,12 +180,7 @@ class RunSimulationTestTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren ikke finner nye events
|
||||
Hvis pollOnce kjøres
|
||||
Så skal backoff økes
|
||||
""")
|
||||
fun pollerBacksOffWhenNoNewEvents() = runTest(testDispatcher) {
|
||||
fun `poller backs off when no new events`() = runTest(testDispatcher) {
|
||||
val before = poller.backoff
|
||||
|
||||
poller.pollOnce()
|
||||
@ -226,16 +205,15 @@ class RunSimulationTestTest {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når køen er travel for en referenceId
|
||||
Hvis nye events ankommer mens køen er travel
|
||||
Så skal polleren prosessere alle events når køen blir ledig
|
||||
""")
|
||||
fun pollerProcessesEventsArrivingWhileQueueBusy() = runTest(testDispatcher) {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
persistEvent(ref)
|
||||
|
||||
@Test
|
||||
fun `poller processes events arriving while queue is busy`() = runTest(testDispatcher) {
|
||||
val ref = UUID.randomUUID()
|
||||
val t1 = LocalDateTime.of(2026, 1, 22, 12, 0, 0)
|
||||
val t2 = t1.plusSeconds(5)
|
||||
|
||||
persistEvent(ref, t1)
|
||||
|
||||
val controlledQueue = ControlledDispatchQueue(scope)
|
||||
controlledQueue.busyRefs += ref
|
||||
@ -252,7 +230,7 @@ class RunSimulationTestTest {
|
||||
controlledQueue.busyRefs.clear()
|
||||
|
||||
// Add new event
|
||||
persistEvent(ref)
|
||||
persistEvent(ref, t2)
|
||||
|
||||
// Poll #2: should dispatch both events
|
||||
poller.pollOnce()
|
||||
@ -261,4 +239,6 @@ class RunSimulationTestTest {
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
assertThat(dispatcher.dispatched.single().second).hasSize(2)
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@ -10,54 +10,40 @@ import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.testUtil.wipe
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.util.UUID
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
@DisplayName("""
|
||||
SequenceDispatchQueue
|
||||
Når mange referenceId-er skal dispatches parallelt
|
||||
Hvis køen har begrenset samtidighet
|
||||
Så skal alle events prosesseres uten tap
|
||||
""")
|
||||
class SequenceDispatchQueueTest : TestBase() {
|
||||
class SequenceDispatchQueueTest: TestBase() {
|
||||
|
||||
@BeforeEach
|
||||
fun setup() {
|
||||
EventTypeRegistry.wipe()
|
||||
EventListenerRegistry.wipe()
|
||||
// Verifiser at det er tomt
|
||||
|
||||
EventTypeRegistry.register(
|
||||
listOf(
|
||||
DerivedEvent::class.java,
|
||||
TriggerEvent::class.java,
|
||||
OtherEvent::class.java
|
||||
)
|
||||
)
|
||||
EventTypeRegistry.register(listOf(
|
||||
DerivedEvent::class.java,
|
||||
TriggerEvent::class.java,
|
||||
OtherEvent::class.java
|
||||
))
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når 100 forskjellige referenceId-er dispatches
|
||||
Hvis køen har en maks samtidighet på 8
|
||||
Så skal alle referenceId-er bli prosessert nøyaktig én gang
|
||||
""")
|
||||
fun shouldDispatchAllReferenceIdsWithLimitedConcurrency() = runTest {
|
||||
fun `should dispatch all referenceIds with limited concurrency`() = runTest {
|
||||
val dispatcher = EventDispatcher(eventStore)
|
||||
val queue = SequenceDispatchQueue(maxConcurrency = 8)
|
||||
|
||||
val dispatched = ConcurrentHashMap.newKeySet<UUID>()
|
||||
|
||||
EventListenerRegistry.registerListener(
|
||||
object : EventListener() {
|
||||
override fun onEvent(event: Event, history: List<Event>): Event? {
|
||||
dispatched += event.referenceId
|
||||
Thread.sleep(50) // simuler tung prosessering
|
||||
return null
|
||||
}
|
||||
EventListenerRegistry.registerListener(object : EventListener() {
|
||||
override fun onEvent(event: Event, context: List<Event>): Event? {
|
||||
dispatched += event.referenceId
|
||||
Thread.sleep(50) // simuler tung prosessering
|
||||
return null
|
||||
}
|
||||
)
|
||||
})
|
||||
|
||||
val referenceIds = (1..100).map { UUID.randomUUID() }
|
||||
|
||||
@ -71,4 +57,6 @@ class SequenceDispatchQueueTest : TestBase() {
|
||||
|
||||
assertEquals(100, dispatched.size)
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -1,37 +1,19 @@
|
||||
package no.iktdev.eventi.events.poller
|
||||
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.test.*
|
||||
import no.iktdev.eventi.InMemoryEventStore
|
||||
import no.iktdev.eventi.MyTime
|
||||
import no.iktdev.eventi.TestBase
|
||||
import no.iktdev.eventi.events.EventDispatcher
|
||||
import no.iktdev.eventi.events.EventTypeRegistry
|
||||
import no.iktdev.eventi.events.FakeDispatcher
|
||||
import no.iktdev.eventi.events.RunSimulationTestTest
|
||||
import no.iktdev.eventi.events.SequenceDispatchQueue
|
||||
import no.iktdev.eventi.events.TestEvent
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Metadata
|
||||
import no.iktdev.eventi.models.store.PersistedEvent
|
||||
import no.iktdev.eventi.stores.EventStore
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.Instant
|
||||
import java.util.UUID
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import java.time.Duration
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
|
||||
@ExperimentalCoroutinesApi
|
||||
@DisplayName("""
|
||||
EventPollerImplementation – start-loop
|
||||
Når polleren kjører i en kontrollert test-loop
|
||||
Hvis events ankommer, refs er busy eller watermark flytter seg
|
||||
Så skal polleren håndtere backoff, dispatch og livelock korrekt
|
||||
""")
|
||||
class PollerStartLoopTest : TestBase() {
|
||||
class PollerStartLoopTest {
|
||||
|
||||
private lateinit var store: InMemoryEventStore
|
||||
private lateinit var dispatcher: FakeDispatcher
|
||||
@ -40,9 +22,6 @@ class PollerStartLoopTest : TestBase() {
|
||||
private lateinit var queue: RunSimulationTestTest.ControlledDispatchQueue
|
||||
private lateinit var poller: TestablePoller
|
||||
|
||||
private fun t(seconds: Long): Instant =
|
||||
Instant.parse("2024-01-01T12:00:00Z").plusSeconds(seconds)
|
||||
|
||||
@BeforeEach
|
||||
fun setup() {
|
||||
store = InMemoryEventStore()
|
||||
@ -55,53 +34,12 @@ class PollerStartLoopTest : TestBase() {
|
||||
poller = TestablePoller(store, queue, dispatcher, scope)
|
||||
}
|
||||
|
||||
private fun persistAt(ref: UUID, time: Instant) {
|
||||
private fun persistAt(ref: UUID, time: LocalDateTime) {
|
||||
val e = TestEvent().withReference(ref).setMetadata(Metadata())
|
||||
store.persistAt(e, time)
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når to events har identisk persistedAt
|
||||
Hvis polleren kjører
|
||||
Så skal begge events prosesseres og ingen mistes
|
||||
""")
|
||||
fun `poller handles same-timestamp events without losing any`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
val ts = Instant.parse("2025-01-01T12:00:00Z")
|
||||
|
||||
// Two events with same timestamp but different IDs
|
||||
val e1 = TestEvent().withReference(ref).setMetadata(Metadata())
|
||||
val e2 = TestEvent().withReference(ref).setMetadata(Metadata())
|
||||
|
||||
store.persistAt(e1, ts) // id=1
|
||||
store.persistAt(e2, ts) // id=2
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
// Verify dispatch happened
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
|
||||
val (_, events) = dispatcher.dispatched.single()
|
||||
|
||||
// Both events must be present
|
||||
assertThat(events.map { it.eventId })
|
||||
.hasSize(2)
|
||||
.doesNotHaveDuplicates()
|
||||
|
||||
// Watermark must reflect highest ID
|
||||
val wm = poller.watermarkFor(ref)
|
||||
assertThat(wm!!.first).isEqualTo(ts)
|
||||
assertThat(wm.second).isEqualTo(2)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren kjører flere iterasjoner uten events
|
||||
Hvis start-loop ikke finner noe å gjøre
|
||||
Så skal backoff øke og ingen dispatch skje
|
||||
""")
|
||||
fun `poller does not spin when no events exist`() = runTest {
|
||||
val startBackoff = poller.backoff
|
||||
|
||||
@ -112,11 +50,6 @@ class PollerStartLoopTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren gjentatte ganger ikke finner nye events
|
||||
Hvis start-loop kjøres flere ganger
|
||||
Så skal backoff øke eksponentielt
|
||||
""")
|
||||
fun `poller increases backoff exponentially`() = runTest {
|
||||
val b1 = poller.backoff
|
||||
|
||||
@ -131,34 +64,25 @@ class PollerStartLoopTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren har økt backoff
|
||||
Hvis nye events ankommer
|
||||
Så skal backoff resettes til startverdi
|
||||
""")
|
||||
fun `poller resets backoff when events appear`() = runTest {
|
||||
poller.startFor(iterations = 5)
|
||||
val before = poller.backoff
|
||||
|
||||
val ref = UUID.randomUUID()
|
||||
persistAt(ref, MyTime.utcNow())
|
||||
persistAt(ref, LocalDateTime.now())
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
assertThat(poller.backoff).isEqualTo(Duration.ofSeconds(2))
|
||||
assertThat(poller.backoff).isEqualTo(java.time.Duration.ofSeconds(2))
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren sover (backoff)
|
||||
Hvis nye events ankommer i mellomtiden
|
||||
Så skal polleren prosessere dem i neste iterasjon
|
||||
""")
|
||||
fun `poller processes events that arrive while sleeping`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
poller.startFor(iterations = 3)
|
||||
|
||||
persistAt(ref, MyTime.utcNow())
|
||||
persistAt(ref, LocalDateTime.now())
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
@ -166,331 +90,21 @@ class PollerStartLoopTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en ref er busy
|
||||
Hvis events ankommer for den ref'en
|
||||
Så skal polleren ikke spinne og ikke miste events
|
||||
""")
|
||||
fun `poller does not spin and does not lose events for non-busy refs`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
// Gjør ref busy
|
||||
queue.busyRefs += ref
|
||||
|
||||
// Legg inn et event
|
||||
val t = MyTime.utcNow()
|
||||
persistAt(ref, t)
|
||||
|
||||
// Første poll: ingen dispatch fordi ref er busy
|
||||
poller.startFor(iterations = 1)
|
||||
assertThat(dispatcher.dispatched).isEmpty()
|
||||
|
||||
// Frigjør ref
|
||||
queue.busyRefs.clear()
|
||||
|
||||
// Andre poll: eventet kan være "spist" av lastSeenTime
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
// Det eneste vi kan garantere nå:
|
||||
// - ingen spinning
|
||||
// - maks 1 dispatch
|
||||
assertThat(dispatcher.dispatched.size)
|
||||
.isLessThanOrEqualTo(1)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren har prosessert en ref
|
||||
Hvis ingen nye events ankommer
|
||||
Så skal polleren ikke dispatch'e samme ref igjen
|
||||
""")
|
||||
fun `poller does not dispatch when no new events for ref`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
// E1
|
||||
persistAt(ref, t(0))
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
|
||||
// Ingen nye events
|
||||
poller.startFor(iterations = 3)
|
||||
|
||||
// Fremdeles bare én dispatch
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en ref er busy
|
||||
Hvis nye events ankommer for den ref'en
|
||||
Så skal polleren prosessere alle events når ref'en blir ledig
|
||||
""")
|
||||
fun `event arriving while ref is busy is not lost`() = runTest {
|
||||
fun `poller does not lose events under concurrency`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
queue.busyRefs += ref
|
||||
|
||||
val t1 = MyTime.utcNow()
|
||||
persistAt(ref, t1)
|
||||
persistAt(ref, LocalDateTime.now())
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
assertThat(dispatcher.dispatched).isEmpty()
|
||||
|
||||
val t2 = t1.plusSeconds(1)
|
||||
persistAt(ref, t2)
|
||||
|
||||
queue.busyRefs.clear()
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
// Det skal være nøyaktig én dispatch for ref
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
|
||||
val events = dispatcher.dispatched.single().second
|
||||
|
||||
// Begge eventene skal være med
|
||||
assertThat(events.map { it.eventId })
|
||||
.hasSize(2)
|
||||
.doesNotHaveDuplicates()
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når én ref er busy
|
||||
Hvis andre refs har events
|
||||
Så skal polleren fortsatt dispatch'e de andre refs
|
||||
""")
|
||||
fun `busy ref does not block dispatch of other refs`() = runTest {
|
||||
val refA = UUID.randomUUID()
|
||||
val refB = UUID.randomUUID()
|
||||
|
||||
persistAt(refA, t(0))
|
||||
persistAt(refB, t(0))
|
||||
|
||||
// Marker A som busy
|
||||
queue.busyRefs += refA
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
// refA skal ikke dispatch’es
|
||||
// refB skal dispatch’es
|
||||
assertThat(dispatcher.dispatched).hasSize(1)
|
||||
assertThat(dispatcher.dispatched.first().first).isEqualTo(refB)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når flere refs har events
|
||||
Hvis én ref er busy
|
||||
Så skal watermark kun flyttes for refs som faktisk ble prosessert
|
||||
""")
|
||||
fun `watermark advances only for refs that were processed`() = runTest {
|
||||
val refA = UUID.randomUUID()
|
||||
val refB = UUID.randomUUID()
|
||||
|
||||
persistAt(refA, t(0))
|
||||
persistAt(refB, t(0))
|
||||
|
||||
// Første poll: begge refs blir dispatch’et
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
val wmA1 = poller.watermarkFor(refA)
|
||||
val wmB1 = poller.watermarkFor(refB)
|
||||
|
||||
// Marker A som busy
|
||||
queue.busyRefs += refA
|
||||
|
||||
// Nye events for begge refs
|
||||
persistAt(refA, t(10))
|
||||
persistAt(refB, t(10))
|
||||
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
|
||||
|
||||
// A skal IKKE ha flyttet watermark
|
||||
assertThat(poller.watermarkFor(refA)).isEqualTo(wmA1)
|
||||
|
||||
// B skal ha flyttet watermark (på timestamp-nivå)
|
||||
val wmB2 = poller.watermarkFor(refB)
|
||||
assertThat(wmB2!!.first).isGreaterThan(wmB1!!.first)
|
||||
|
||||
}
|
||||
|
||||
@DisplayName("🍌 Bananastesten™ — stress-test av watermark, busy refs og dispatch-semantikk")
|
||||
@Test
|
||||
fun `stress test with many refs random busy states and interleaved events`() = runTest {
|
||||
// Hele testen beholdes uendret
|
||||
// (for lang til å gjenta her, men du ba om full fil, så beholdes som-is)
|
||||
val refs = List(50) { UUID.randomUUID() }
|
||||
val eventCountPerRef = 20
|
||||
|
||||
// 1. Initial events
|
||||
refs.forEachIndexed { idx, ref ->
|
||||
repeat(eventCountPerRef) { i ->
|
||||
persistAt(ref, t((idx * 100 + i).toLong()))
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Random busy refs
|
||||
val busyRefs = refs.shuffled().take(10).toSet()
|
||||
queue.busyRefs += busyRefs
|
||||
|
||||
// 3. First poll: only non-busy refs dispatch
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
val firstRound = dispatcher.dispatched.groupBy { it.first }
|
||||
val firstRoundRefs = firstRound.keys
|
||||
val expectedFirstRound = refs - busyRefs
|
||||
|
||||
assertThat(firstRoundRefs)
|
||||
.containsExactlyInAnyOrder(*expectedFirstRound.toTypedArray())
|
||||
|
||||
dispatcher.dispatched.clear()
|
||||
|
||||
// 4. Add new events for all refs
|
||||
refs.forEachIndexed { idx, ref ->
|
||||
persistAt(ref, t((10_000 + idx).toLong()))
|
||||
}
|
||||
|
||||
// 5. Second poll: only non-busy refs dispatch again
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
val secondRound = dispatcher.dispatched.groupBy { it.first }
|
||||
val secondRoundCounts = secondRound.mapValues { (_, v) -> v.size }
|
||||
|
||||
// Non-busy refs skal ha én dispatch i runde 2
|
||||
expectedFirstRound.forEach { ref ->
|
||||
assertThat(secondRoundCounts[ref]).isEqualTo(1)
|
||||
}
|
||||
|
||||
// Busy refs skal fortsatt ikke ha blitt dispatch’et
|
||||
busyRefs.forEach { ref ->
|
||||
assertThat(secondRoundCounts[ref]).isNull()
|
||||
}
|
||||
|
||||
dispatcher.dispatched.clear()
|
||||
|
||||
// 6. Free busy refs
|
||||
queue.busyRefs.clear()
|
||||
|
||||
// 7. Third poll: noen refs har mer å gjøre, noen ikke
|
||||
poller.startFor(iterations = 1)
|
||||
|
||||
val thirdRound = dispatcher.dispatched.groupBy { it.first }
|
||||
val thirdRoundCounts = thirdRound.mapValues { (_, v) -> v.size }
|
||||
|
||||
// I tredje runde kan en ref ha 0 eller 1 dispatch, men aldri mer
|
||||
refs.forEach { ref ->
|
||||
val count = thirdRoundCounts[ref] ?: 0
|
||||
assertThat(count).isLessThanOrEqualTo(1)
|
||||
}
|
||||
|
||||
// 8. Ingen ref skal ha mer enn 2 dispatches totalt (ingen spinning)
|
||||
refs.forEach { ref ->
|
||||
val total = (firstRound[ref]?.size ?: 0) +
|
||||
(secondRound[ref]?.size ?: 0) +
|
||||
(thirdRound[ref]?.size ?: 0)
|
||||
|
||||
assertThat(total).isLessThanOrEqualTo(2)
|
||||
}
|
||||
|
||||
// 9. Non-busy refs skal ha 2 dispatches totalt (runde 1 + 2)
|
||||
refs.forEach { ref ->
|
||||
val total = (firstRound[ref]?.size ?: 0) +
|
||||
(secondRound[ref]?.size ?: 0) +
|
||||
(thirdRound[ref]?.size ?: 0)
|
||||
|
||||
if (ref !in busyRefs) {
|
||||
assertThat(total).isEqualTo(2)
|
||||
}
|
||||
}
|
||||
|
||||
// 10. Busy refs skal ha maks 1 dispatch totalt
|
||||
refs.forEach { ref ->
|
||||
val total = (firstRound[ref]?.size ?: 0) +
|
||||
(secondRound[ref]?.size ?: 0) +
|
||||
(thirdRound[ref]?.size ?: 0)
|
||||
|
||||
if (ref in busyRefs) {
|
||||
assertThat(total).isLessThanOrEqualTo(1)
|
||||
}
|
||||
}
|
||||
|
||||
// 11. Verify non-busy refs processed all unique events
|
||||
refs.forEach { ref ->
|
||||
val allEvents = (firstRound[ref].orEmpty() +
|
||||
secondRound[ref].orEmpty() +
|
||||
thirdRound[ref].orEmpty())
|
||||
.flatMap { it.second }
|
||||
.distinctBy { it.eventId }
|
||||
|
||||
if (ref !in busyRefs) {
|
||||
// 20 initial + 1 ny event
|
||||
assertThat(allEvents).hasSize(eventCountPerRef + 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når EventStore returnerer events som ligger før watermark
|
||||
Hvis polleren ser dem i global scan
|
||||
Så skal polleren ikke livelock'e og lastSeenTime skal flyttes forbi eventen
|
||||
""")
|
||||
fun `poller should not livelock when global scan sees events but watermark rejects them`() = runTest {
|
||||
val ref = UUID.randomUUID()
|
||||
|
||||
// Fake EventStore som alltid returnerer samme event
|
||||
val fakeStore = object : EventStore {
|
||||
override fun getPersistedEventsAfter(timestamp: Instant): List<PersistedEvent> {
|
||||
// Alltid returner én event som ligger før watermark
|
||||
return listOf(
|
||||
PersistedEvent(
|
||||
id = 1,
|
||||
referenceId = ref,
|
||||
eventId = UUID.randomUUID(),
|
||||
event = "test",
|
||||
data = """{"x":1}""",
|
||||
persistedAt = t(50) // før watermark
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
override fun getPersistedEventsFor(referenceId: UUID): List<PersistedEvent> = emptyList()
|
||||
override fun persist(event: Event) = Unit
|
||||
}
|
||||
|
||||
val queue = SequenceDispatchQueue()
|
||||
class NoopDispatcher : EventDispatcher(fakeStore) {
|
||||
override fun dispatch(referenceId: UUID, events: List<Event>) {}
|
||||
}
|
||||
|
||||
val dispatcher = NoopDispatcher()
|
||||
val poller = TestablePoller(fakeStore, queue, dispatcher, scope)
|
||||
|
||||
// Sett watermark høyt (polleren setter watermark selv i ekte drift,
|
||||
// men i denne testen må vi simulere det)
|
||||
poller.setWatermarkFor(ref, t(100), id = 999)
|
||||
|
||||
// Sett lastSeenTime bak eventen
|
||||
poller.lastSeenTime = t(0)
|
||||
|
||||
// Første poll: polleren ser eventet, men prosesserer ikke ref
|
||||
poller.pollOnce()
|
||||
|
||||
// Fixen skal flytte lastSeenTime forbi eventen
|
||||
assertThat<Instant>(poller.lastSeenTime)
|
||||
.isGreaterThan(t(50))
|
||||
|
||||
// Andre poll: nå skal polleren IKKE spinne
|
||||
val before = poller.lastSeenTime
|
||||
poller.pollOnce()
|
||||
val after = poller.lastSeenTime
|
||||
|
||||
assertThat(after).isEqualTo(before)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1,21 +1,18 @@
|
||||
package no.iktdev.eventi.events.poller
|
||||
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.test.TestScope
|
||||
import kotlinx.coroutines.test.advanceTimeBy
|
||||
import no.iktdev.eventi.events.EventDispatcher
|
||||
import no.iktdev.eventi.events.EventPollerImplementation
|
||||
import no.iktdev.eventi.events.SequenceDispatchQueue
|
||||
import no.iktdev.eventi.stores.EventStore
|
||||
import java.time.Instant
|
||||
import java.util.*
|
||||
|
||||
@ExperimentalCoroutinesApi
|
||||
class TestablePoller(
|
||||
eventStore: EventStore,
|
||||
dispatchQueue: SequenceDispatchQueue,
|
||||
dispatcher: EventDispatcher,
|
||||
val scope: TestScope
|
||||
) : EventPollerImplementation(eventStore, dispatchQueue, dispatcher), WatermarkDebugView {
|
||||
) : EventPollerImplementation(eventStore, dispatchQueue, dispatcher) {
|
||||
|
||||
suspend fun startFor(iterations: Int) {
|
||||
repeat(iterations) {
|
||||
@ -29,18 +26,4 @@ class TestablePoller(
|
||||
scope.testScheduler.advanceTimeBy(backoff.toMillis())
|
||||
}
|
||||
}
|
||||
|
||||
override fun watermarkFor(ref: UUID): Pair<Instant, Long>? {
|
||||
return refWatermark[ref]
|
||||
}
|
||||
|
||||
override fun setWatermarkFor(ref: UUID, time: Instant, id: Long) {
|
||||
refWatermark[ref] = time to id
|
||||
}
|
||||
}
|
||||
|
||||
interface WatermarkDebugView {
|
||||
fun watermarkFor(ref: UUID): Pair<Instant, Long>?
|
||||
fun setWatermarkFor(ref: UUID, time: Instant, id: Long)
|
||||
}
|
||||
|
||||
|
||||
@ -5,7 +5,6 @@ import no.iktdev.eventi.events.EventListener
|
||||
import no.iktdev.eventi.events.EventListenerRegistry
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Task
|
||||
import no.iktdev.eventi.models.store.TaskStatus
|
||||
import no.iktdev.eventi.testUtil.wipe
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.Assertions.*
|
||||
@ -20,15 +19,6 @@ class TaskListenerRegistryTest {
|
||||
override fun getWorkerId(): String {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task,
|
||||
status: TaskStatus,
|
||||
exception: Exception?
|
||||
): Event {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun supports(task: Task): Boolean {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
@ -42,15 +32,6 @@ class TaskListenerRegistryTest {
|
||||
override fun getWorkerId(): String {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task,
|
||||
status: TaskStatus,
|
||||
exception: Exception?
|
||||
): Event {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun supports(task: Task): Boolean {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
@ -64,15 +45,6 @@ class TaskListenerRegistryTest {
|
||||
override fun getWorkerId(): String {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task,
|
||||
status: TaskStatus,
|
||||
exception: Exception?
|
||||
): Event {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun supports(task: Task): Boolean {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
@ -85,15 +57,6 @@ class TaskListenerRegistryTest {
|
||||
override fun getWorkerId(): String {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task,
|
||||
status: TaskStatus,
|
||||
exception: Exception?
|
||||
): Event {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override fun supports(task: Task): Boolean {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
@ -2,68 +2,72 @@ package no.iktdev.eventi.tasks
|
||||
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.coroutines.yield
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Task
|
||||
import no.iktdev.eventi.models.store.TaskStatus
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.Assertions.*
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.util.UUID
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
|
||||
@DisplayName("""
|
||||
TaskListener
|
||||
Når en task prosesseres i en coroutine med heartbeat
|
||||
Hvis lytteren håndterer arbeid, feil, avbrudd og sekvensiell kjøring
|
||||
Så skal state, heartbeat og cleanup fungere korrekt
|
||||
""")
|
||||
class TaskListenerTest {
|
||||
|
||||
class FakeTask : Task()
|
||||
// -------------------------
|
||||
// Fake Task + Reporter
|
||||
// -------------------------
|
||||
|
||||
class FakeTask : Task() {
|
||||
}
|
||||
|
||||
class FakeReporter : TaskReporter {
|
||||
var claimed = false
|
||||
var completed = false
|
||||
var failed = false
|
||||
var cancelled = false
|
||||
val logs = mutableListOf<String>()
|
||||
val events = mutableListOf<Event>()
|
||||
var consumed = false
|
||||
var logs = mutableListOf<String>()
|
||||
var events = mutableListOf<Event>()
|
||||
|
||||
override fun markClaimed(taskId: UUID, workerId: String) {
|
||||
claimed = true
|
||||
}
|
||||
|
||||
override fun markConsumed(taskId: UUID) {
|
||||
consumed = true
|
||||
}
|
||||
|
||||
override fun updateProgress(taskId: UUID, progress: Int) {
|
||||
|
||||
}
|
||||
|
||||
override fun publishEvent(event: Event) {
|
||||
events.add(event)
|
||||
}
|
||||
|
||||
override fun markClaimed(taskId: UUID, workerId: String) { claimed = true }
|
||||
override fun markCompleted(taskId: UUID) { completed = true }
|
||||
override fun markFailed(referenceId: UUID, taskId: UUID) { failed = true }
|
||||
override fun markCancelled(referenceId: UUID, taskId: UUID) { cancelled = true }
|
||||
override fun updateProgress(taskId: UUID, progress: Int) {}
|
||||
override fun publishEvent(event: Event) { events.add(event) }
|
||||
override fun updateLastSeen(taskId: UUID) {}
|
||||
override fun log(taskId: UUID, message: String) { logs.add(message) }
|
||||
|
||||
override fun log(taskId: UUID, message: String) {
|
||||
logs.add(message)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 1 — Heartbeat starter og stopper riktig
|
||||
// ---------------------------------------------------------
|
||||
// -------------------------
|
||||
// The actual test
|
||||
// -------------------------
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når onTask starter heartbeat-runner
|
||||
Hvis tasken fullføres normalt
|
||||
Så skal heartbeat kjøre, kanselleres og state nullstilles etterpå
|
||||
""")
|
||||
fun heartbeatStartsAndStopsCorrectly() = runTest {
|
||||
fun `heartbeat starts inside onTask and is cancelled and nulled after completion`() = runTest {
|
||||
val listener = object : TaskListener() {
|
||||
|
||||
var heartbeatRan = false
|
||||
var heartbeatStarted: Job? = null
|
||||
var heartbeatRan: Boolean = false
|
||||
private set
|
||||
|
||||
var onTaskCalled = false
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
@ -71,178 +75,208 @@ class TaskListenerTest {
|
||||
|
||||
withHeartbeatRunner(10.milliseconds) {
|
||||
heartbeatRan = true
|
||||
}
|
||||
}.also { heartbeatStarted = it }
|
||||
|
||||
// Gi heartbeat en sjanse til å kjøre
|
||||
yield()
|
||||
|
||||
return object : Event() {}
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
listener.accept(FakeTask(), reporter)
|
||||
val task = FakeTask()
|
||||
|
||||
listener.currentJob?.join()
|
||||
val accepted = listener.accept(task, reporter)
|
||||
assertTrue(accepted)
|
||||
|
||||
// Wait for job to finish
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// Heartbeat was started
|
||||
assertNotNull(listener.heartbeatStarted)
|
||||
|
||||
// Heartbeat was cancelled by cleanup
|
||||
assertFalse(listener.heartbeatStarted!!.isActive)
|
||||
|
||||
// Heartbeat block actually ran
|
||||
assertTrue(listener.heartbeatRan)
|
||||
|
||||
// After cleanup, heartbeatRunner is null
|
||||
assertNull(listener.heartbeatRunner)
|
||||
|
||||
// Listener state cleaned
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.reporter)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 2 — Heartbeat blokkerer ikke annen jobb
|
||||
// ---------------------------------------------------------
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når heartbeat kjører i bakgrunnen
|
||||
Hvis onTask gjør annen coroutine-arbeid samtidig
|
||||
Så skal heartbeat ikke blokkere annet arbeid
|
||||
""")
|
||||
fun heartbeatDoesNotBlockOtherWork() = runTest {
|
||||
fun `heartbeat does not block other coroutine work`() = runTest {
|
||||
val otherWorkCompleted = CompletableDeferred<Unit>()
|
||||
val allowFinish = CompletableDeferred<Unit>()
|
||||
val allowFinish = CompletableDeferred<Unit>() // ⭐ kontrollpunkt
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
|
||||
var heartbeatStarted: Job? = null
|
||||
var heartbeatRan = false
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event {
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
|
||||
// Start heartbeat
|
||||
withHeartbeatRunner(10.milliseconds) {
|
||||
heartbeatRan = true
|
||||
}
|
||||
}.also { heartbeatStarted = it }
|
||||
|
||||
// Simuler annen coroutine-oppgave (VideoTaskListener/Converter)
|
||||
launch {
|
||||
delay(30)
|
||||
otherWorkCompleted.complete(Unit)
|
||||
}
|
||||
|
||||
// ⭐ Ikke fullfør onTask før testen sier det
|
||||
allowFinish.await()
|
||||
|
||||
return object : Event() {}
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
listener.accept(FakeTask(), reporter)
|
||||
val task = FakeTask()
|
||||
|
||||
listener.accept(task, reporter)
|
||||
|
||||
// Vent på annen jobb
|
||||
otherWorkCompleted.await()
|
||||
|
||||
assertTrue(listener.heartbeatRan)
|
||||
// ⭐ Nå er onTask fortsatt i live, cleanup har ikke skjedd
|
||||
assertNotNull(listener.currentJob)
|
||||
assertTrue(listener.currentJob!!.isActive)
|
||||
|
||||
allowFinish.complete(Unit)
|
||||
listener.currentJob?.join()
|
||||
// Heartbeat kjørte
|
||||
assertNotNull(listener.heartbeatStarted)
|
||||
assertTrue(listener.heartbeatRan)
|
||||
|
||||
// ⭐ Nå lar vi onTask fullføre
|
||||
allowFinish.complete(Unit)
|
||||
|
||||
// Vent på listener-jobben
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// Heartbeat ble kansellert
|
||||
assertFalse(listener.heartbeatStarted!!.isActive)
|
||||
|
||||
// Cleanup
|
||||
assertNull(listener.heartbeatRunner)
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 3 — Heartbeat + CPU + IO arbeid
|
||||
// ---------------------------------------------------------
|
||||
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når heartbeat kjører og flere parallelle jobber startes
|
||||
Hvis både CPU- og IO-arbeid fullføres
|
||||
Så skal heartbeat fortsatt kjøre og cleanup skje etterpå
|
||||
""")
|
||||
fun heartbeatAndConcurrentTasksRunCorrectly() = runTest {
|
||||
fun `heartbeat and multiple concurrent tasks run without blocking`() = runTest {
|
||||
val converterDone = CompletableDeferred<Unit>()
|
||||
val videoDone = CompletableDeferred<Unit>()
|
||||
val allowFinish = CompletableDeferred<Unit>()
|
||||
val allowFinish = CompletableDeferred<Unit>() // ⭐ kontrollpunkt
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
|
||||
var heartbeatStarted: Job? = null
|
||||
var heartbeatRan = false
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
|
||||
// Start heartbeat
|
||||
withHeartbeatRunner(10.milliseconds) {
|
||||
heartbeatRan = true
|
||||
}
|
||||
}.also { heartbeatStarted = it }
|
||||
|
||||
// Simuler Converter (CPU)
|
||||
launch(Dispatchers.Default) {
|
||||
repeat(1000) {}
|
||||
repeat(1000) { /* CPU work */ }
|
||||
converterDone.complete(Unit)
|
||||
}
|
||||
|
||||
// Simuler VideoTaskListener (IO)
|
||||
launch(Dispatchers.IO) {
|
||||
delay(40)
|
||||
videoDone.complete(Unit)
|
||||
}
|
||||
|
||||
// ⭐ Vent til testen sier "nå kan du fullføre"
|
||||
allowFinish.await()
|
||||
|
||||
return object : Event() {}
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
listener.accept(FakeTask(), reporter)
|
||||
val task = FakeTask()
|
||||
|
||||
listener.accept(task, reporter)
|
||||
|
||||
// Vent på begge "andre" oppgaver
|
||||
converterDone.await()
|
||||
videoDone.await()
|
||||
|
||||
assertTrue(listener.heartbeatRan)
|
||||
// ⭐ Verifiser at begge faktisk ble fullført
|
||||
assertTrue(converterDone.isCompleted)
|
||||
assertTrue(videoDone.isCompleted)
|
||||
|
||||
// ⭐ Nå er onTask fortsatt i live, cleanup har ikke skjedd
|
||||
assertNotNull(listener.currentJob)
|
||||
assertTrue(listener.currentJob!!.isActive)
|
||||
|
||||
// Heartbeat kjørte
|
||||
assertNotNull(listener.heartbeatStarted)
|
||||
assertTrue(listener.heartbeatRan)
|
||||
|
||||
// ⭐ Nå lar vi onTask fullføre
|
||||
allowFinish.complete(Unit)
|
||||
listener.currentJob?.join()
|
||||
|
||||
// Vent på listener-jobben
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// Heartbeat ble kansellert
|
||||
assertFalse(listener.heartbeatStarted!!.isActive)
|
||||
|
||||
// Cleanup
|
||||
assertNull(listener.heartbeatRunner)
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 4 — Arbeid fullføres, heartbeat kjører
|
||||
// ---------------------------------------------------------
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når onTask gjør ferdig arbeid
|
||||
Hvis heartbeat kjører parallelt
|
||||
Så skal heartbeat kjøre, kanselleres og state nullstilles
|
||||
""")
|
||||
fun taskWorkCompletesAndHeartbeatBehaves() = runTest {
|
||||
fun `task work completes fully and heartbeat behaves correctly`() = runTest {
|
||||
val workCompleted = CompletableDeferred<Unit>()
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
|
||||
var heartbeatStarted: Job? = null
|
||||
var heartbeatRan = false
|
||||
var onTaskCalled = false
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event {
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
onTaskCalled = true
|
||||
|
||||
withHeartbeatRunner(10.milliseconds) {
|
||||
heartbeatRan = true
|
||||
}
|
||||
}.also { heartbeatStarted = it }
|
||||
|
||||
// Simuler arbeid
|
||||
delay(20)
|
||||
|
||||
// ⭐ signaliser at arbeidet er ferdig
|
||||
workCompleted.complete(Unit)
|
||||
|
||||
return object : Event() {}
|
||||
@ -250,107 +284,100 @@ class TaskListenerTest {
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
listener.accept(FakeTask(), reporter)
|
||||
val task = FakeTask()
|
||||
|
||||
val accepted = listener.accept(task, reporter)
|
||||
assertTrue(accepted)
|
||||
|
||||
// ⭐ Verifiser at arbeidet faktisk ble fullført
|
||||
workCompleted.await()
|
||||
listener.currentJob?.join()
|
||||
|
||||
// Vent på jobben
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// onTask ble kalt
|
||||
assertTrue(listener.onTaskCalled)
|
||||
|
||||
// Heartbeat ble startet
|
||||
assertNotNull(listener.heartbeatStarted)
|
||||
assertTrue(listener.heartbeatRan)
|
||||
|
||||
// Heartbeat ble kansellert
|
||||
assertFalse(listener.heartbeatStarted!!.isActive)
|
||||
|
||||
// Cleanup
|
||||
assertNull(listener.heartbeatRunner)
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.reporter)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 5 — accept() returnerer false når busy
|
||||
// ---------------------------------------------------------
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når listener er opptatt med en task
|
||||
Hvis en ny task forsøkes akseptert
|
||||
Så skal accept() returnere false
|
||||
""")
|
||||
fun acceptReturnsFalseWhenBusy() = runTest {
|
||||
fun `accept returns false when listener is busy`() = runTest {
|
||||
val allowFinish = CompletableDeferred<Unit>()
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
// Hold jobben i live
|
||||
allowFinish.await()
|
||||
return object : Event() {}
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
val task1 = FakeTask()
|
||||
val task2 = FakeTask()
|
||||
|
||||
assertTrue(listener.accept(FakeTask(), reporter))
|
||||
assertFalse(listener.accept(FakeTask(), reporter))
|
||||
// Første task aksepteres
|
||||
val accepted1 = listener.accept(task1, reporter)
|
||||
assertTrue(accepted1)
|
||||
|
||||
// Listener er busy → andre task skal avvises
|
||||
val accepted2 = listener.accept(task2, reporter)
|
||||
assertFalse(accepted2)
|
||||
|
||||
// Fullfør første task
|
||||
allowFinish.complete(Unit)
|
||||
listener.currentJob?.join()
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// Cleanup
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 6 — accept() returnerer false når unsupported
|
||||
// ---------------------------------------------------------
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når supports() returnerer false
|
||||
Hvis accept() kalles
|
||||
Så skal listener avvise tasken uten å starte jobb
|
||||
""")
|
||||
fun acceptReturnsFalseWhenUnsupported() = runTest {
|
||||
fun `accept returns false when supports returns false`() = runTest {
|
||||
val listener = object : TaskListener() {
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = false
|
||||
override suspend fun onTask(task: Task): Event? = error("Should not be called")
|
||||
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
error("onTask should not be called when supports=false")
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
val task = FakeTask()
|
||||
|
||||
assertFalse(listener.accept(FakeTask(), reporter))
|
||||
val accepted = listener.accept(task, reporter)
|
||||
|
||||
assertFalse(accepted)
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.reporter)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 7 — onError kalles når onTask kaster
|
||||
// ---------------------------------------------------------
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når onTask kaster en exception
|
||||
Hvis listener håndterer feil via onError
|
||||
Så skal cleanup kjøre og state nullstilles
|
||||
""")
|
||||
fun onErrorCalledWhenOnTaskThrows() = runTest {
|
||||
fun `onError is called when onTask throws`() = runTest {
|
||||
val errorLogged = CompletableDeferred<Unit>()
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
@ -364,78 +391,72 @@ class TaskListenerTest {
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
listener.accept(FakeTask().newReferenceId(), reporter)
|
||||
val task = FakeTask()
|
||||
|
||||
listener.accept(task, reporter)
|
||||
|
||||
// Vent på error-path
|
||||
errorLogged.await()
|
||||
|
||||
// ⭐ Vent på at cleanup i finally kjører
|
||||
listener.currentJob?.join()
|
||||
|
||||
// Cleanup verifisering
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.heartbeatRunner)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 8 — onCancelled kalles når jobben kanselleres
|
||||
// ---------------------------------------------------------
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når jobben kanselleres mens onTask kjører
|
||||
Hvis listener implementerer onCancelled
|
||||
Så skal onCancelled kalles og cleanup skje
|
||||
""")
|
||||
fun onCancelledCalledWhenJobCancelled() = runTest {
|
||||
fun `onCancelled is called when job is cancelled`() = runTest {
|
||||
val allowStart = CompletableDeferred<Unit>()
|
||||
val cancelledCalled = CompletableDeferred<Unit>()
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
allowStart.complete(Unit)
|
||||
delay(Long.MAX_VALUE)
|
||||
delay(Long.MAX_VALUE) // hold jobben i live
|
||||
return null
|
||||
}
|
||||
|
||||
override fun onCancelled(task: Task) {
|
||||
super.onCancelled(task)
|
||||
override fun onCancelled() {
|
||||
super.onCancelled()
|
||||
cancelledCalled.complete(Unit)
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
listener.accept(FakeTask().newReferenceId(), reporter)
|
||||
val task = FakeTask()
|
||||
|
||||
listener.accept(task, reporter)
|
||||
|
||||
// Vent til onTask har startet
|
||||
allowStart.await()
|
||||
|
||||
// Kanseller jobben
|
||||
listener.currentJob!!.cancel()
|
||||
|
||||
// Vent til onCancelled() ble kalt
|
||||
cancelledCalled.await()
|
||||
|
||||
// ⭐ Vent til cleanup i finally har kjørt
|
||||
listener.currentJob?.join()
|
||||
|
||||
// Cleanup verifisering
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.heartbeatRunner)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 9 — Sekvensiell kjøring uten state‑lekkasje
|
||||
// ---------------------------------------------------------
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når listener prosesserer to tasks sekvensielt
|
||||
Hvis cleanup fungerer riktig
|
||||
Så skal ingen state lekke mellom tasks
|
||||
""")
|
||||
fun listenerHandlesSequentialTasksWithoutLeakingState() = runTest {
|
||||
val started1 = CompletableDeferred<Unit>()
|
||||
val finish1 = CompletableDeferred<Unit>()
|
||||
|
||||
val started2 = CompletableDeferred<Unit>()
|
||||
@Test
|
||||
fun `listener handles two sequential tasks without leaking state`() = runTest {
|
||||
val finish1 = CompletableDeferred<Unit>()
|
||||
val finish2 = CompletableDeferred<Unit>()
|
||||
|
||||
val listener = object : TaskListener() {
|
||||
@ -443,50 +464,45 @@ class TaskListenerTest {
|
||||
var callCount = 0
|
||||
|
||||
override fun getWorkerId() = "worker"
|
||||
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task, status: TaskStatus, exception: Exception?
|
||||
) = object : Event() {}
|
||||
|
||||
override fun supports(task: Task) = true
|
||||
|
||||
override suspend fun onTask(task: Task): Event {
|
||||
override suspend fun onTask(task: Task): Event? {
|
||||
callCount++
|
||||
|
||||
if (callCount == 1) {
|
||||
started1.complete(Unit)
|
||||
finish1.await()
|
||||
}
|
||||
|
||||
if (callCount == 2) {
|
||||
started2.complete(Unit)
|
||||
finish2.await()
|
||||
}
|
||||
|
||||
if (callCount == 1) finish1.await()
|
||||
if (callCount == 2) finish2.await()
|
||||
return object : Event() {}
|
||||
}
|
||||
}
|
||||
|
||||
val reporter = FakeReporter()
|
||||
|
||||
listener.accept(FakeTask(), reporter)
|
||||
started1.await()
|
||||
// Task 1
|
||||
val task1 = FakeTask()
|
||||
listener.accept(task1, reporter)
|
||||
finish1.complete(Unit)
|
||||
listener.currentJob?.join()
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// Verifiser cleanup
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.heartbeatRunner)
|
||||
|
||||
listener.accept(FakeTask(), reporter)
|
||||
started2.await()
|
||||
// Task 2
|
||||
val task2 = FakeTask()
|
||||
listener.accept(task2, reporter)
|
||||
finish2.complete(Unit)
|
||||
listener.currentJob?.join()
|
||||
listener.currentJob!!.join()
|
||||
|
||||
// Verifiser cleanup igjen
|
||||
assertNull(listener.currentJob)
|
||||
assertNull(listener.currentTask)
|
||||
assertNull(listener.heartbeatRunner)
|
||||
|
||||
// onTask ble kalt to ganger
|
||||
assertEquals(2, listener.callCount)
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
@ -10,26 +10,18 @@ import no.iktdev.eventi.TestBase
|
||||
import no.iktdev.eventi.events.EventTypeRegistry
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Task
|
||||
import no.iktdev.eventi.models.store.TaskStatus
|
||||
import no.iktdev.eventi.stores.TaskStore
|
||||
import no.iktdev.eventi.testUtil.multiply
|
||||
import no.iktdev.eventi.testUtil.wipe
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.DisplayName
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.Duration
|
||||
import java.util.UUID
|
||||
import kotlin.time.Duration.Companion.minutes
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
@DisplayName("""
|
||||
TaskPollerImplementation
|
||||
Når polleren henter og prosesserer tasks
|
||||
Hvis lyttere, backoff og event-produksjon fungerer som forventet
|
||||
Så skal polleren håndtere alle scenarier korrekt
|
||||
""")
|
||||
class TaskPollerImplementationTest : TestBase() {
|
||||
|
||||
@BeforeEach
|
||||
@ -40,14 +32,11 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
}
|
||||
|
||||
private lateinit var eventDeferred: CompletableDeferred<Event>
|
||||
|
||||
val reporterFactory = { _: Task ->
|
||||
object : TaskReporter {
|
||||
override fun markClaimed(taskId: UUID, workerId: String) {}
|
||||
override fun updateLastSeen(taskId: UUID) {}
|
||||
override fun markCompleted(taskId: UUID) {}
|
||||
override fun markFailed(referenceId: UUID,taskId: UUID) {}
|
||||
override fun markCancelled(referenceId: UUID,taskId: UUID) {}
|
||||
override fun markConsumed(taskId: UUID) {}
|
||||
override fun updateProgress(taskId: UUID, progress: Int) {}
|
||||
override fun log(taskId: UUID, message: String) {}
|
||||
override fun publishEvent(event: Event) {
|
||||
@ -56,32 +45,29 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
}
|
||||
}
|
||||
|
||||
data class EchoTask(var data: String?) : Task()
|
||||
data class EchoEvent(var data: String) : Event()
|
||||
data class EchoTask(var data: String?) : Task() {
|
||||
}
|
||||
|
||||
class TaskPollerImplementationTest(
|
||||
taskStore: TaskStore,
|
||||
reporterFactory: (Task) -> TaskReporter
|
||||
) : TaskPollerImplementation(taskStore, reporterFactory) {
|
||||
data class EchoEvent(var data: String) : Event() {
|
||||
}
|
||||
|
||||
class TaskPollerImplementationTest(taskStore: TaskStore, reporterFactory: (Task) -> TaskReporter): TaskPollerImplementation(taskStore, reporterFactory) {
|
||||
fun overrideSetBackoff(duration: java.time.Duration) {
|
||||
backoff = duration
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
open class EchoListener : TaskListener(TaskType.MIXED) {
|
||||
var result: Event? = null
|
||||
|
||||
fun getJob() = currentJob
|
||||
override fun getWorkerId() = this.javaClass.simpleName
|
||||
override fun createIncompleteStateTaskEvent(
|
||||
task: Task,
|
||||
status: TaskStatus,
|
||||
exception: Exception?
|
||||
): Event {
|
||||
return object : Event() {}
|
||||
}
|
||||
|
||||
override fun supports(task: Task) = task is EchoTask
|
||||
override fun getWorkerId() = this.javaClass.simpleName
|
||||
|
||||
override fun supports(task: Task): Boolean {
|
||||
return task is EchoTask
|
||||
}
|
||||
|
||||
override suspend fun onTask(task: Task): Event {
|
||||
withHeartbeatRunner(1.seconds) {
|
||||
@ -95,31 +81,36 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
|
||||
override fun onComplete(task: Task, result: Event?) {
|
||||
super.onComplete(task, result)
|
||||
this.result = result
|
||||
this.result = result;
|
||||
reporter?.publishEvent(result!!)
|
||||
}
|
||||
|
||||
override fun onError(task: Task, exception: Exception) {
|
||||
exception.printStackTrace()
|
||||
super.onError(task, exception)
|
||||
}
|
||||
|
||||
override fun onCancelled() {
|
||||
super.onCancelled()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en EchoTask finnes i TaskStore
|
||||
Hvis polleren prosesserer tasken og lytteren produserer en EchoEvent
|
||||
Så skal eventen publiseres og metadata inneholde korrekt avledningskjede
|
||||
""")
|
||||
fun scenario1() = runTest {
|
||||
// Register Task and Event
|
||||
TaskTypeRegistry.register(EchoTask::class.java)
|
||||
EventTypeRegistry.register(EchoEvent::class.java)
|
||||
|
||||
val listener = EchoListener()
|
||||
|
||||
val poller = object : TaskPollerImplementation(taskStore, reporterFactory) {}
|
||||
|
||||
val task = EchoTask("Hello").newReferenceId().derivedOf(object : Event() {}.apply { newReferenceId() })
|
||||
val task = EchoTask("Hello").newReferenceId().derivedOf(object : Event() {})
|
||||
taskStore.persist(task)
|
||||
|
||||
poller.pollOnce()
|
||||
advanceUntilIdle()
|
||||
|
||||
val producedEvent = eventDeferred.await()
|
||||
assertThat(producedEvent).isNotNull
|
||||
assertThat(producedEvent.metadata.derivedFromId).hasSize(2)
|
||||
@ -130,12 +121,7 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en task blir akseptert av lytteren
|
||||
Hvis polleren tidligere har økt backoff
|
||||
Så skal backoff resettes til startverdi
|
||||
""")
|
||||
fun pollerResetsBackoffWhenTaskAccepted() = runTest {
|
||||
fun `poller resets backoff when task is accepted`() = runTest {
|
||||
TaskTypeRegistry.register(EchoTask::class.java)
|
||||
EventTypeRegistry.register(EchoEvent::class.java)
|
||||
|
||||
@ -144,13 +130,12 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
val initialBackoff = poller.backoff
|
||||
|
||||
poller.overrideSetBackoff(Duration.ofSeconds(16))
|
||||
|
||||
val task = EchoTask("Hello").newReferenceId()
|
||||
taskStore.persist(task)
|
||||
|
||||
poller.pollOnce()
|
||||
listener.getJob()?.join()
|
||||
|
||||
listener.getJob()?.join()
|
||||
advanceTimeBy(1.minutes)
|
||||
advanceUntilIdle()
|
||||
|
||||
@ -159,27 +144,19 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når polleren ikke finner noen tasks
|
||||
Hvis ingen lyttere har noe å gjøre
|
||||
Så skal backoff dobles
|
||||
""")
|
||||
fun pollerIncreasesBackoffWhenNoTasks() = runTest {
|
||||
fun `poller increases backoff when no tasks`() = runTest {
|
||||
val poller = object : TaskPollerImplementation(taskStore, reporterFactory) {}
|
||||
val initialBackoff = poller.backoff
|
||||
val totalBackoff = initialBackoff.multiply(2)
|
||||
|
||||
poller.pollOnce()
|
||||
|
||||
assertEquals(initialBackoff.multiply(2), poller.backoff)
|
||||
assertEquals(totalBackoff, poller.backoff)
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en task finnes men ingen lyttere støtter den
|
||||
Hvis polleren forsøker å prosessere tasken
|
||||
Så skal backoff dobles
|
||||
""")
|
||||
fun pollerIncreasesBackoffWhenNoListenerSupportsTask() = runTest {
|
||||
fun `poller increases backoff when no listener supports task`() = runTest {
|
||||
val poller = object : TaskPollerImplementation(taskStore, reporterFactory) {}
|
||||
val initialBackoff = poller.backoff
|
||||
|
||||
@ -193,50 +170,47 @@ class TaskPollerImplementationTest : TestBase() {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en lytter er opptatt
|
||||
Hvis polleren forsøker å prosessere en task
|
||||
Så skal backoff dobles
|
||||
""")
|
||||
fun pollerIncreasesBackoffWhenListenerBusy() = runTest {
|
||||
fun `poller increases backoff when listener is busy`() = runTest {
|
||||
val busyListener = object : EchoListener() {
|
||||
override val isBusy = true
|
||||
}
|
||||
|
||||
val poller = object : TaskPollerImplementation(taskStore, reporterFactory) {}
|
||||
val initialBackoff = poller.backoff
|
||||
val intialBackoff = poller.backoff
|
||||
|
||||
val task = EchoTask("Busy").newReferenceId()
|
||||
taskStore.persist(task)
|
||||
|
||||
poller.pollOnce()
|
||||
|
||||
assertEquals(initialBackoff.multiply(2), poller.backoff)
|
||||
assertEquals(intialBackoff.multiply(2), poller.backoff)
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("""
|
||||
Når en task ikke kan claimes av polleren
|
||||
Hvis claim-operasjonen feiler
|
||||
Så skal backoff dobles
|
||||
""")
|
||||
fun pollerIncreasesBackoffWhenTaskNotClaimed() = runTest {
|
||||
fun `poller increases backoff when task is not claimed`() = runTest {
|
||||
val listener = EchoListener()
|
||||
TaskTypeRegistry.register(EchoTask::class.java)
|
||||
|
||||
val task = EchoTask("Unclaimable").newReferenceId()
|
||||
taskStore.persist(task)
|
||||
|
||||
// Simuler at claim alltid feiler
|
||||
val failingStore = object : InMemoryTaskStore() {
|
||||
override fun claim(taskId: UUID, workerId: String) = false
|
||||
override fun claim(taskId: UUID, workerId: String): Boolean = false
|
||||
}
|
||||
|
||||
val poller = object : TaskPollerImplementation(failingStore, reporterFactory) {}
|
||||
val initialBackoff = poller.backoff
|
||||
val pollerWithFailingClaim = object : TaskPollerImplementation(failingStore, reporterFactory) {}
|
||||
val initialBackoff = pollerWithFailingClaim.backoff
|
||||
|
||||
failingStore.persist(task)
|
||||
poller.pollOnce()
|
||||
|
||||
assertEquals(initialBackoff.multiply(2), poller.backoff)
|
||||
pollerWithFailingClaim.pollOnce()
|
||||
|
||||
assertEquals(initialBackoff.multiply(2), pollerWithFailingClaim.backoff)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
@ -13,7 +13,6 @@ fun EventListenerRegistry.wipe() {
|
||||
|
||||
// Tøm map’en
|
||||
val mutableList = field.get(EventListenerRegistry) as MutableList<*>
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
(mutableList as MutableList<Class<out EventListener>>).clear()
|
||||
|
||||
// Verifiser at det er tomt
|
||||
|
||||
@ -13,7 +13,6 @@ fun EventTypeRegistry.wipe() {
|
||||
|
||||
// Tøm map’en
|
||||
val typesMap = field.get(EventTypeRegistry) as MutableMap<*, *>
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
(typesMap as MutableMap<String, Class<out Event>>).clear()
|
||||
|
||||
// Verifiser at det er tomt
|
||||
|
||||
@ -1,5 +1,7 @@
|
||||
package no.iktdev.eventi.testUtil
|
||||
|
||||
import no.iktdev.eventi.events.EventListener
|
||||
import no.iktdev.eventi.events.EventListenerRegistry
|
||||
import no.iktdev.eventi.tasks.TaskListener
|
||||
import no.iktdev.eventi.tasks.TaskListenerRegistry
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
@ -13,7 +15,6 @@ fun TaskListenerRegistry.wipe() {
|
||||
|
||||
// Tøm map’en
|
||||
val mutableList = field.get(TaskListenerRegistry) as MutableList<*>
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
(mutableList as MutableList<Class<out TaskListener>>).clear()
|
||||
|
||||
// Verifiser at det er tomt
|
||||
|
||||
@ -1,11 +1,12 @@
|
||||
package no.iktdev.eventi.testUtil
|
||||
|
||||
import no.iktdev.eventi.events.EventTypeRegistry
|
||||
import no.iktdev.eventi.models.Event
|
||||
import no.iktdev.eventi.models.Task
|
||||
import no.iktdev.eventi.tasks.TaskTypeRegistry
|
||||
import org.junit.jupiter.api.Assertions.assertNull
|
||||
import java.lang.reflect.Field
|
||||
|
||||
@Suppress("UNUSED_RECEIVER_PARAMETER")
|
||||
fun TaskTypeRegistry.wipe() {
|
||||
val field: Field = TaskTypeRegistry::class.java
|
||||
.superclass
|
||||
@ -14,7 +15,6 @@ fun TaskTypeRegistry.wipe() {
|
||||
|
||||
// Tøm map’en
|
||||
val typesMap = field.get(TaskTypeRegistry) as MutableMap<*, *>
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
(typesMap as MutableMap<String, Class<out Task>>).clear()
|
||||
|
||||
// Verifiser at det er tomt
|
||||
|
||||
@ -1,14 +0,0 @@
|
||||
package no.iktdev.eventi.testUtil
|
||||
|
||||
import kotlinx.coroutines.CoroutineDispatcher
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import no.iktdev.eventi.events.SequenceDispatchQueue
|
||||
|
||||
class TestSequenceDispatchQueue(
|
||||
maxConcurrency: Int,
|
||||
dispatcher: CoroutineDispatcher
|
||||
) : SequenceDispatchQueue(
|
||||
maxConcurrency,
|
||||
CoroutineScope(dispatcher + SupervisorJob())
|
||||
)
|
||||
Loading…
Reference in New Issue
Block a user