feat: priorityを設定できるように

broker/broker-mongodb/src/main/kotlin/dev/usbharu/owl/broker/mongodb/MongoModuleContext.kt

@@ -36,7 +36,7 @@ class MongoModuleContext : ModuleContext {
                             ConnectionString(
                                 System.getProperty(
                                     "owl.broker.mongo.url",
-                                    "mongodb://agent1.build:27017"
+                                    "mongodb://localhost:27017"
                                 )
                             )
                         )

broker/broker-mongodb/src/main/kotlin/dev/usbharu/owl/broker/mongodb/MongodbQueuedTaskRepository.kt

@@ -109,6 +109,7 @@ data class QueuedTaskMongodb(
     val task: TaskMongodb,
     val attempt: Int,
     val queuedAt: Instant,
+    val priority:Int,
     val isActive: Boolean,
     val timeoutAt: Instant?,
     val assignedConsumer: String?,
@@ -117,13 +118,14 @@ data class QueuedTaskMongodb(
 
     fun toQueuedTask(propertySerializerFactory: PropertySerializerFactory): QueuedTask {
         return QueuedTask(
-            attempt,
-            queuedAt,
-            task.toTask(propertySerializerFactory),
-            isActive,
-            timeoutAt,
-            assignedConsumer?.let { UUID.fromString(it) },
-            assignedAt
+            attempt = attempt,
+            queuedAt = queuedAt,
+            task = task.toTask(propertySerializerFactory),
+            priority = priority,
+            isActive = isActive,
+            timeoutAt = timeoutAt,
+            assignedConsumer = assignedConsumer?.let { UUID.fromString(it) },
+            assignedAt = assignedAt
         )
     }
 
@@ -174,6 +176,7 @@ data class QueuedTaskMongodb(
                 TaskMongodb.of(propertySerializerFactory, queuedTask.task),
                 queuedTask.attempt,
                 queuedTask.queuedAt,
+                queuedTask.priority,
                 queuedTask.isActive,
                 queuedTask.timeoutAt,
                 queuedTask.assignedConsumer?.toString(),

broker/src/main/kotlin/dev/usbharu/owl/broker/domain/model/queuedtask/QueuedTask.kt

@@ -28,6 +28,7 @@ data class QueuedTask(
     val attempt: Int,
     val queuedAt: Instant,
     val task: Task,
+    val priority: Int,
     val isActive: Boolean,
     val timeoutAt: Instant?,
     val assignedConsumer: UUID?,

broker/src/main/kotlin/dev/usbharu/owl/broker/interfaces/grpc/TaskResultSubscribeService.kt

@@ -16,18 +16,14 @@
 
 package dev.usbharu.owl.broker.interfaces.grpc
 
-import TaskResultProducer.TaskResults
-import TaskResultSubscribeServiceGrpcKt
-import dev.usbharu.owl.Uuid
+import dev.usbharu.owl.*
 import dev.usbharu.owl.broker.external.toUUID
 import dev.usbharu.owl.broker.service.TaskManagementService
 import dev.usbharu.owl.common.property.PropertySerializeUtils
 import dev.usbharu.owl.common.property.PropertySerializerFactory
-import dev.usbharu.owl.taskResult
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 import org.koin.core.annotation.Singleton
-import taskResults
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
 
@@ -38,7 +34,7 @@ class TaskResultSubscribeService(
     coroutineContext: CoroutineContext = EmptyCoroutineContext
 ) :
     TaskResultSubscribeServiceGrpcKt.TaskResultSubscribeServiceCoroutineImplBase(coroutineContext) {
-    override fun subscribe(request: Uuid.UUID): Flow<TaskResults> {
+    override fun subscribe(request: Uuid.UUID): Flow<TaskResultProducer.TaskResults> {
         return taskManagementService
             .subscribeResult(request.toUUID())
             .map {

broker/src/main/kotlin/dev/usbharu/owl/broker/service/TaskManagementService.kt

@@ -83,18 +83,20 @@ class TaskManagementServiceImpl(
 
     private suspend fun enqueueTask(task: Task): QueuedTask {
 
-        val queuedTask = QueuedTask(
-            task.attempt + 1,
-            Instant.now(),
-            task,
-            isActive = true,
-            timeoutAt = null,
-            null,
-            null
-        )
-
         val definedTask = taskDefinitionRepository.findByName(task.name)
             ?: throw TaskNotRegisterException("Task ${task.name} not definition.")
+
+        val queuedTask = QueuedTask(
+            attempt = task.attempt + 1,
+            queuedAt = Instant.now(),
+            task = task,
+            priority = definedTask.priority,
+            isActive = true,
+            timeoutAt = null,
+            assignedConsumer = null,
+            assignedAt = null
+        )
+
         val copy = task.copy(
             nextRetry = retryPolicyFactory.factory(definedTask.retryPolicy)
                 .nextRetry(Instant.now(), queuedTask.attempt)

broker/src/main/proto/task_result_producer.proto

@@ -2,6 +2,8 @@ syntax = "proto3";
 import "uuid.proto";
 import "task_result.proto";
 
+option java_package = "dev.usbharu.owl";
+
 message TaskResults {
   string name = 1;
   UUID id = 2;