feat: MongoDBへのアクセスをIOスレッドで行うように

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

@@ -21,7 +21,9 @@ import com.mongodb.client.model.ReplaceOptions
 import com.mongodb.kotlin.client.coroutine.MongoDatabase
 import dev.usbharu.owl.broker.domain.model.consumer.Consumer
 import dev.usbharu.owl.broker.domain.model.consumer.ConsumerRepository
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.singleOrNull
+import kotlinx.coroutines.withContext
 import org.bson.BsonType
 import org.bson.codecs.pojo.annotations.BsonId
 import org.bson.codecs.pojo.annotations.BsonRepresentation
@@ -32,13 +34,13 @@ import java.util.*
 class MongodbConsumerRepository(database: MongoDatabase) : ConsumerRepository {
 
     private val collection = database.getCollection<ConsumerMongodb>("consumers")
-    override suspend fun save(consumer: Consumer): Consumer {
+    override suspend fun save(consumer: Consumer): Consumer = withContext(Dispatchers.IO) {
         collection.replaceOne(Filters.eq("_id", consumer.id.toString()), ConsumerMongodb.of(consumer), ReplaceOptions().upsert(true))
-        return consumer
+        return@withContext consumer
     }
 
-    override suspend fun findById(id: UUID): Consumer? {
+    override suspend fun findById(id: UUID): Consumer? = withContext(Dispatchers.IO) {
-        return collection.find(Filters.eq("_id", id.toString())).singleOrNull()?.toConsumer()
+        return@withContext collection.find(Filters.eq("_id", id.toString())).singleOrNull()?.toConsumer()
     }
 }
 

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

@@ -21,6 +21,8 @@ import com.mongodb.client.model.ReplaceOptions
 import com.mongodb.kotlin.client.coroutine.MongoDatabase
 import dev.usbharu.owl.broker.domain.model.producer.Producer
 import dev.usbharu.owl.broker.domain.model.producer.ProducerRepository
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import org.koin.core.annotation.Singleton
 import java.time.Instant
 import java.util.*
@@ -30,13 +32,13 @@ class MongodbProducerRepository(database: MongoDatabase) : ProducerRepository {
 
     private val collection = database.getCollection<ProducerMongodb>("producers")
 
-    override suspend fun save(producer: Producer): Producer {
+    override suspend fun save(producer: Producer): Producer = withContext(Dispatchers.IO) {
         collection.replaceOne(
             Filters.eq("_id", producer.id.toString()),
             ProducerMongodb.of(producer),
             ReplaceOptions().upsert(true)
         )
-        return producer
+        return@withContext producer
     }
 }
 

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

@@ -16,7 +16,6 @@
 
 package dev.usbharu.owl.broker.mongodb
 
-import com.mongodb.client.model.Filters
 import com.mongodb.client.model.Filters.*
 import com.mongodb.client.model.FindOneAndUpdateOptions
 import com.mongodb.client.model.ReplaceOptions
@@ -28,8 +27,11 @@ import dev.usbharu.owl.broker.domain.model.queuedtask.QueuedTaskRepository
 import dev.usbharu.owl.broker.domain.model.task.Task
 import dev.usbharu.owl.common.property.PropertySerializeUtils
 import dev.usbharu.owl.common.property.PropertySerializerFactory
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.withContext
 import org.bson.BsonType
 import org.bson.codecs.pojo.annotations.BsonId
 import org.bson.codecs.pojo.annotations.BsonRepresentation
@@ -45,14 +47,18 @@ class MongodbQueuedTaskRepository(
 
     private val collection = database.getCollection<QueuedTaskMongodb>("queued_task")
     override suspend fun save(queuedTask: QueuedTask): QueuedTask {
+        withContext(Dispatchers.IO) {
             collection.replaceOne(
                 eq("_id", queuedTask.task.id.toString()), QueuedTaskMongodb.of(propertySerializerFactory, queuedTask),
                 ReplaceOptions().upsert(true)
             )
+        }
         return queuedTask
     }
 
     override suspend fun findByTaskIdAndAssignedConsumerIsNullAndUpdate(id: UUID, update: QueuedTask): QueuedTask {
+        return withContext(Dispatchers.IO) {
+
             val findOneAndUpdate = collection.findOneAndUpdate(
                 and(
                     eq("_id", id.toString()),
@@ -67,7 +73,8 @@ class MongodbQueuedTaskRepository(
             if (findOneAndUpdate == null) {
                 TODO()
             }
-        return findOneAndUpdate.toQueuedTask(propertySerializerFactory)
+            findOneAndUpdate.toQueuedTask(propertySerializerFactory)
+        }
     }
 
     override fun findByTaskNameInAndAssignedConsumerIsNullAndOrderByPriority(
@@ -79,12 +86,12 @@ class MongodbQueuedTaskRepository(
                 `in`("task.name", tasks),
                 eq(QueuedTaskMongodb::assignedConsumer.name, null)
             )
-        ).map { it.toQueuedTask(propertySerializerFactory) }
+        ).map { it.toQueuedTask(propertySerializerFactory) }.flowOn(Dispatchers.IO)
     }
 
     override fun findByQueuedAtBeforeAndAssignedConsumerIsNull(instant: Instant): Flow<QueuedTask> {
-        return collection.find(Filters.lte(QueuedTaskMongodb::queuedAt.name, instant))
+        return collection.find(lte(QueuedTaskMongodb::queuedAt.name, instant))
-            .map { it.toQueuedTask(propertySerializerFactory) }
+            .map { it.toQueuedTask(propertySerializerFactory) }.flowOn(Dispatchers.IO)
     }
 }
 

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

@@ -21,7 +21,9 @@ import com.mongodb.client.model.ReplaceOptions
 import com.mongodb.kotlin.client.coroutine.MongoDatabase
 import dev.usbharu.owl.broker.domain.model.taskdefinition.TaskDefinition
 import dev.usbharu.owl.broker.domain.model.taskdefinition.TaskDefinitionRepository
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.singleOrNull
+import kotlinx.coroutines.withContext
 import org.bson.BsonType
 import org.bson.codecs.pojo.annotations.BsonId
 import org.bson.codecs.pojo.annotations.BsonRepresentation
@@ -31,21 +33,21 @@ import org.koin.core.annotation.Singleton
 class MongodbTaskDefinitionRepository(database: MongoDatabase) : TaskDefinitionRepository {
 
     private val collection = database.getCollection<TaskDefinitionMongodb>("task_definition")
-    override suspend fun save(taskDefinition: TaskDefinition): TaskDefinition {
+    override suspend fun save(taskDefinition: TaskDefinition): TaskDefinition = withContext(Dispatchers.IO) {
         collection.replaceOne(
             Filters.eq("_id", taskDefinition.name),
             TaskDefinitionMongodb.of(taskDefinition),
             ReplaceOptions().upsert(true)
         )
-        return taskDefinition
+        return@withContext taskDefinition
     }
 
-    override suspend fun deleteByName(name: String) {
+    override suspend fun deleteByName(name: String): Unit = withContext(Dispatchers.IO) {
         collection.deleteOne(Filters.eq("_id",name))
     }
 
-    override suspend fun findByName(name: String): TaskDefinition? {
+    override suspend fun findByName(name: String): TaskDefinition? = withContext(Dispatchers.IO) {
-        return collection.find(Filters.eq("_id", name)).singleOrNull()?.toTaskDefinition()
+        return@withContext collection.find(Filters.eq("_id", name)).singleOrNull()?.toTaskDefinition()
     }
 }
 

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

@@ -23,8 +23,11 @@ import dev.usbharu.owl.broker.domain.model.task.Task
 import dev.usbharu.owl.broker.domain.model.task.TaskRepository
 import dev.usbharu.owl.common.property.PropertySerializeUtils
 import dev.usbharu.owl.common.property.PropertySerializerFactory
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.withContext
 import org.bson.BsonType
 import org.bson.codecs.pojo.annotations.BsonId
 import org.bson.codecs.pojo.annotations.BsonRepresentation
@@ -37,17 +40,17 @@ class MongodbTaskRepository(database: MongoDatabase, private val propertySeriali
     TaskRepository {
 
     private val collection = database.getCollection<TaskMongodb>("tasks")
-    override suspend fun save(task: Task): Task {
+    override suspend fun save(task: Task): Task = withContext(Dispatchers.IO) {
         collection.replaceOne(
             Filters.eq("_id", task.id.toString()), TaskMongodb.of(propertySerializerFactory, task),
             ReplaceOptions().upsert(true)
         )
-        return task
+        return@withContext task
     }
 
     override fun findByNextRetryBefore(timestamp: Instant): Flow<Task> {
         return collection.find(Filters.lte(TaskMongodb::nextRetry.name, timestamp))
-            .map { it.toTask(propertySerializerFactory) }
+            .map { it.toTask(propertySerializerFactory) }.flowOn(Dispatchers.IO)
     }
 }