feat: タスクをまとめて追加できるように

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

@@ -17,6 +17,7 @@
 package dev.usbharu.owl.broker.mongodb
 
 import com.mongodb.client.model.Filters
+import com.mongodb.client.model.ReplaceOneModel
 import com.mongodb.client.model.ReplaceOptions
 import com.mongodb.kotlin.client.coroutine.MongoDatabase
 import dev.usbharu.owl.broker.domain.model.task.Task
@@ -49,6 +50,16 @@ class MongodbTaskRepository(database: MongoDatabase, private val propertySeriali
         return@withContext task
     }
 
+    override suspend fun saveAll(tasks: List<Task>): Unit = withContext(Dispatchers.IO) {
+        collection.bulkWrite(tasks.map {
+            ReplaceOneModel(
+                Filters.eq(it.id.toString()),
+                TaskMongodb.of(propertySerializerFactory, it),
+                ReplaceOptions().upsert(true)
+            )
+        })
+    }
+
     override fun findByNextRetryBefore(timestamp: Instant): Flow<Task> {
         return collection.find(Filters.lte(TaskMongodb::nextRetry.name, timestamp))
             .map { it.toTask(propertySerializerFactory) }.flowOn(Dispatchers.IO)

broker/src/main/kotlin/dev/usbharu/owl/broker/domain/model/task/TaskRepository.kt

@@ -23,6 +23,8 @@ import java.util.*
 interface TaskRepository {
     suspend fun save(task: Task):Task
 
+    suspend fun saveAll(tasks:List<Task>)
+
     fun findByNextRetryBefore(timestamp:Instant): Flow<Task>
 
     suspend fun findById(uuid: UUID): Task?

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

@@ -18,6 +18,7 @@ package dev.usbharu.owl.broker.interfaces.grpc
 
 import dev.usbharu.owl.PublishTaskOuterClass
 import dev.usbharu.owl.PublishTaskOuterClass.PublishedTask
+import dev.usbharu.owl.PublishTaskOuterClass.PublishedTasks
 import dev.usbharu.owl.TaskPublishServiceGrpcKt.TaskPublishServiceCoroutineImplBase
 import dev.usbharu.owl.broker.external.toUUID
 import dev.usbharu.owl.broker.service.PublishTask
@@ -56,14 +57,28 @@ class TaskPublishService(
             )
             PublishedTask.newBuilder().setName(publishedTask.name).setId(publishedTask.id.toUUID()).build()
         } catch (e: Throwable) {
-            logger.warn("exception ",e)
+            logger.warn("exception ", e)
             throw StatusException(Status.INTERNAL)
         }
-
-
     }
 
-    companion object{
+    override suspend fun publishTasks(request: PublishTaskOuterClass.PublishTasks): PublishTaskOuterClass.PublishedTasks {
-        private val logger = LoggerFactory.getLogger(dev.usbharu.owl.broker.interfaces.grpc.TaskPublishService::class.java)
+
+        val tasks = request.propertiesArrayList.map {
+            PublishTask(
+                request.name,
+                request.producerId.toUUID(),
+                PropertySerializeUtils.deserialize(propertySerializerFactory, it.propertiesMap)
+            )
+        }
+
+        val publishTasks = taskPublishService.publishTasks(tasks)
+
+        return PublishedTasks.newBuilder().setName(request.name).addAllId(publishTasks.map { it.id.toUUID() }).build()
+    }
+
+    companion object {
+        private val logger =
+            LoggerFactory.getLogger(dev.usbharu.owl.broker.interfaces.grpc.TaskPublishService::class.java)
     }
 }

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

@@ -28,6 +28,7 @@ import java.util.*
 
 interface TaskPublishService {
     suspend fun publishTask(publishTask: PublishTask): PublishedTask
+    suspend fun publishTasks(list: List<PublishTask>): List<PublishedTask>
 }
 
 data class PublishTask(
@@ -44,11 +45,11 @@ data class PublishedTask(
 @Singleton
 class TaskPublishServiceImpl(
     private val taskRepository: TaskRepository,
-    private val taskDefinitionRepository:TaskDefinitionRepository,
+    private val taskDefinitionRepository: TaskDefinitionRepository,
     private val retryPolicyFactory: RetryPolicyFactory
 ) : TaskPublishService {
     override suspend fun publishTask(publishTask: PublishTask): PublishedTask {
-                val id = UUID.randomUUID()
+        val id = UUID.randomUUID()
 
         val definition = taskDefinitionRepository.findByName(publishTask.name)
             ?: throw TaskNotRegisterException("Task ${publishTask.name} not definition.")
@@ -77,6 +78,37 @@ class TaskPublishServiceImpl(
         )
     }
 
+    override suspend fun publishTasks(list: List<PublishTask>): List<PublishedTask> {
+
+        val first = list.first()
+
+        val definition = taskDefinitionRepository.findByName(first.name)
+            ?: throw TaskNotRegisterException("Task ${first.name} not definition.")
+
+        val published = Instant.now()
+
+        val nextRetry = retryPolicyFactory.factory(definition.retryPolicy).nextRetry(published, 0)
+
+        val tasks = list.map {
+            Task(
+                it.name,
+                UUID.randomUUID(),
+                first.producerId,
+                published,
+                nextRetry,
+                null,
+                0,
+                it.properties
+            )
+        }
+
+        taskRepository.saveAll(tasks)
+
+        logger.debug("Published {} tasks. name: {}", tasks.size, first.name)
+
+        return tasks.map { PublishedTask(it.name, it.id) }
+    }
+
     companion object {
         private val logger = LoggerFactory.getLogger(TaskPublishServiceImpl::class.java)
     }

broker/src/main/proto/publish_task.proto

@@ -15,11 +15,28 @@ message PublishTask {
   UUID producer_id = 4;
 }
 
+message Properties {
+  map<string,string> properties = 1;
+}
+
+message PublishTasks {
+  string name = 1;
+  google.protobuf.Timestamp publishedAt = 2;
+  repeated Properties propertiesArray = 3;
+  UUID producer_id = 4;
+}
+
 message PublishedTask {
   string name = 1;
   UUID id = 2;
 }
 
+message PublishedTasks {
+  string name = 1;
+  repeated UUID id = 2;
+}
+
 service TaskPublishService {
   rpc publishTask (PublishTask) returns (PublishedTask);
+  rpc publishTasks(PublishTasks) returns (PublishedTasks);
 }