iktdev/eventi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 40 Wiki Activity

Changes to tasks

Browse Source
This commit is contained in:
Brage Skjønborg 2025-10-12 19:41:38 +02:00
parent 479d4cc25e
commit 597adb36bc
2 changed files with 17 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/main/kotlin/no/iktdev/eventi/tasks/TaskListener.kt
View File

@ -15,7 +15,7 @@ import kotlin.coroutines.cancellation.CancellationException
* @param T The type of result produced by processing the task. * @param T The type of result produced by processing the task.
* @param reporter An instance of [TaskReporter] for reporting task status and events. * @param reporter An instance of [TaskReporter] for reporting task status and events.
*/ */
abstract class TaskListener<T>(val taskType: TaskType = TaskType.CPU_INTENSIVE): TaskListenerImplementation<T> { abstract class TaskListener<T>(val taskType: TaskType = TaskType.CPU_INTENSIVE): TaskListenerImplementation {
init { init {
TaskListenerRegistry.registerListener(this) TaskListenerRegistry.registerListener(this)
@ -69,9 +69,12 @@ abstract class TaskListener<T>(val taskType: TaskType = TaskType.CPU_INTENSIVE):
reporter?.markConsumed(task.taskId) reporter?.markConsumed(task.taskId)
} }
override fun onComplete(task: Task, result: T?) { override fun onComplete(task: Task, result: Event?) {
reporter?.markConsumed(task.taskId) reporter!!.markConsumed(task.taskId)
reporter?.log(task.taskId, "Task completed successfully.") reporter!!.log(task.taskId, "Task completed successfully.")
result?.let {
reporter!!.publishEvent(result)
}
} }
override fun onCancelled() { override fun onCancelled() {
@ -88,11 +91,11 @@ enum class TaskType {
} }
interface TaskListenerImplementation<T> { interface TaskListenerImplementation {
fun supports(task: Task): Boolean fun supports(task: Task): Boolean
fun accept(task: Task, reporter: TaskReporter): Boolean fun accept(task: Task, reporter: TaskReporter): Boolean
suspend fun onTask(task: Task): T suspend fun onTask(task: Task): Event?
fun onComplete(task: Task, result: T?) fun onComplete(task: Task, result: Event?)
fun onError(task: Task, exception: Exception) fun onError(task: Task, exception: Exception)
fun onCancelled() fun onCancelled()
} }

14
src/test/kotlin/no/iktdev/eventi/tasks/AbstractTaskPollerTest.kt
View File

@ -58,7 +58,7 @@ class AbstractTaskPollerTest : TestBase() {
open class EchoListener : TaskListener<String>(TaskType.MIXED) { open class EchoListener : TaskListener<String>(TaskType.MIXED) {
var result: String? = null var result: Event? = null
override fun getWorkerId() = this.javaClass.simpleName override fun getWorkerId() = this.javaClass.simpleName
@ -66,17 +66,17 @@ class AbstractTaskPollerTest : TestBase() {
return task is EchoTask return task is EchoTask
} }
override suspend fun onTask(task: Task): String { override suspend fun onTask(task: Task): Event {
if (task is EchoTask) { if (task is EchoTask) {
return task.data + " Potetmos" return EchoEvent(task.data + " Potetmos").producedFrom(task)
} }
throw IllegalArgumentException("Unsupported task type: ${task::class.java}") throw IllegalArgumentException("Unsupported task type: ${task::class.java}")
} }
override fun onComplete(task: Task, result: String?) { override fun onComplete(task: Task, result: Event?) {
super.onComplete(task, result) super.onComplete(task, result)
this.result = result; this.result = result;
reporter?.publishEvent(EchoEvent(result!!).producedFrom(task)) reporter?.publishEvent(result!!)
} }
} }
@ -99,7 +99,7 @@ class AbstractTaskPollerTest : TestBase() {
val producedEvent = eventDeferred.await() val producedEvent = eventDeferred.await()
assertThat(producedEvent).isNotNull assertThat(producedEvent).isNotNull
assertThat(producedEvent!!.metadata.derivedFromId).isEqualTo(task.taskId) assertThat(producedEvent!!.metadata.derivedFromId).isEqualTo(task.taskId)
assertThat(listener.result).isEqualTo("Hello Potetmos") assertThat((listener.result as EchoEvent).data).isEqualTo("Hello Potetmos")
} }
@OptIn(ExperimentalCoroutinesApi::class) @OptIn(ExperimentalCoroutinesApi::class)
@ -120,7 +120,7 @@ class AbstractTaskPollerTest : TestBase() {
advanceUntilIdle() advanceUntilIdle()
assertEquals(initialBackoff, poller.backoff) assertEquals(initialBackoff, poller.backoff)
assertEquals("Hello Potetmos", listener.result) assertEquals("Hello Potetmos", (listener.result as EchoEvent).data)
} }
@Test @Test