Extract 'BlockingBridge' for IO operations (#1933)

Author: nulls

save-backend/src/main/kotlin/com/saveourtool/save/backend/configs/ApplicationConfiguration.kt

@@ -2,6 +2,7 @@ package com.saveourtool.save.backend.configs
 
 import com.saveourtool.save.service.LogService
 import com.saveourtool.save.service.LokiLogService
+import com.saveourtool.save.utils.BlockingBridge
 import org.springframework.boot.actuate.autoconfigure.metrics.orm.jpa.HibernateMetricsAutoConfiguration
 import org.springframework.boot.autoconfigure.ImportAutoConfiguration
 import org.springframework.boot.autoconfigure.domain.EntityScan
@@ -23,4 +24,10 @@ class ApplicationConfiguration {
      */
     @Bean
     fun logService(configProperties: ConfigProperties): LogService = LokiLogService.createOrStub(configProperties.loki)
+
+    /**
+     * @return [BlockingBridge]
+     */
+    @Bean
+    fun blockingBridge(): BlockingBridge = BlockingBridge.default
 }

save-backend/src/main/kotlin/com/saveourtool/save/backend/storage/FileS3KeyManager.kt

@@ -10,6 +10,7 @@ import com.saveourtool.save.entities.Project
 import com.saveourtool.save.entities.toEntity
 import com.saveourtool.save.storage.concatS3Key
 import com.saveourtool.save.storage.key.AbstractS3KeyDtoManager
+import com.saveourtool.save.utils.BlockingBridge
 import com.saveourtool.save.utils.orNotFound
 
 import org.springframework.data.repository.findByIdOrNull
@@ -24,11 +25,13 @@ import kotlinx.datetime.toJavaLocalDateTime
 class FileS3KeyManager(
     configProperties: ConfigProperties,
     fileRepository: FileRepository,
+    blockingBridge: BlockingBridge,
     private val projectService: ProjectService,
     private val executionService: ExecutionService,
 ) : AbstractS3KeyDtoManager<FileDto, File, FileRepository>(
     concatS3Key(configProperties.s3Storage.prefix, "storage"),
     fileRepository,
+    blockingBridge,
 ) {
     override fun createNewEntityFromDto(dto: FileDto): File = dto.toEntity {
         projectService.findByNameAndOrganizationNameAndCreatedStatus(dto.projectCoordinates.projectName, dto.projectCoordinates.organizationName)

save-backend/src/main/kotlin/com/saveourtool/save/backend/storage/TestsSourceSnapshotS3KeyManager.kt

@@ -11,7 +11,9 @@ import com.saveourtool.save.entities.TestsSourceSnapshot.Companion.toEntity
 import com.saveourtool.save.storage.concatS3Key
 import com.saveourtool.save.storage.key.AbstractS3KeyDtoManager
 import com.saveourtool.save.test.TestsSourceSnapshotDto
+import com.saveourtool.save.utils.BlockingBridge
 import com.saveourtool.save.utils.getByIdOrNotFound
+
 import org.springframework.stereotype.Component
 
 /**
@@ -21,12 +23,14 @@ import org.springframework.stereotype.Component
 class TestsSourceSnapshotS3KeyManager(
     configProperties: ConfigProperties,
     testsSourceSnapshotRepository: TestsSourceSnapshotRepository,
+    blockingBridge: BlockingBridge,
     private val testSuitesSourceRepository: TestSuitesSourceRepository,
     private val testSuitesService: TestSuitesService,
     private val executionService: ExecutionService,
 ) : AbstractS3KeyDtoManager<TestsSourceSnapshotDto, TestsSourceSnapshot, TestsSourceSnapshotRepository>(
     concatS3Key(configProperties.s3Storage.prefix, "tests-source-snapshot"),
     testsSourceSnapshotRepository,
+    blockingBridge
 ) {
     override fun createNewEntityFromDto(dto: TestsSourceSnapshotDto): TestsSourceSnapshot = dto.toEntity { testSuitesSourceRepository.getByIdOrNotFound(it) }
 

save-backend/src/main/resources/application-dev.properties

@@ -9,3 +9,4 @@ backend.s3-storage.presigned-endpoint=http://host.docker.internal:9000
 backend.s3-storage.bucketName=cnb
 backend.s3-storage.credentials.accessKeyId=admin
 backend.s3-storage.credentials.secretAccessKey=adminadmin
+backend.s3-storage.createBucketIfNotExists=true

save-cloud-common/src/jvmMain/kotlin/com/saveourtool/save/storage/DefaultStorageCoroutines.kt

@@ -23,8 +23,6 @@ import kotlinx.coroutines.future.asDeferred
 import kotlinx.coroutines.reactive.asFlow
 import kotlinx.coroutines.reactive.asPublisher
 
-import kotlinx.coroutines.withContext
-
 /**
  * S3 implementation of [StorageCoroutines]
  *
@@ -171,7 +169,7 @@ class DefaultStorageCoroutines<K : Any>(
 
     private suspend fun <R> S3KeyManager<K>.callAsSuspend(function: S3KeyManager<K>.() -> R): R =
             if (s3KeyManager is AbstractS3KeyDatabaseManager<*, *, *>) {
-                withContext(s3KeyManager.ioDispatcher) {
+                s3KeyManager.blockingBridge.blockingToSuspend {
                     function(this@callAsSuspend)
                 }
             } else {

save-cloud-common/src/jvmMain/kotlin/com/saveourtool/save/storage/DefaultStorageProjectReactor.kt

@@ -156,13 +156,10 @@ class DefaultStorageProjectReactor<K : Any>(
 
     private fun createNewS3Key(key: K): Mono<String> = s3KeyManager.callAsMono { createNewS3Key(key) }
 
-    private fun <R : Any> S3KeyManager<K>.callAsMono(function: S3KeyManager<K>.() -> R?): Mono<R> = { function(this) }
-        .toMono()
-        .let {
+    private fun <R : Any> S3KeyManager<K>.callAsMono(function: S3KeyManager<K>.() -> R?): Mono<R> =
             if (s3KeyManager is AbstractS3KeyDatabaseManager<*, *, *>) {
-                it.subscribeOn(s3KeyManager.ioScheduler)
+                s3KeyManager.blockingBridge.blockingToMono { function(this) }
             } else {
-                it
+                { function(this) }.toMono()
             }
-        }
 }

save-cloud-common/src/jvmMain/kotlin/com/saveourtool/save/storage/key/AbstractS3KeyDatabaseManager.kt

@@ -2,29 +2,23 @@ package com.saveourtool.save.storage.key
 
 import com.saveourtool.save.spring.entity.BaseEntity
 import com.saveourtool.save.spring.repository.BaseEntityRepository
+import com.saveourtool.save.utils.BlockingBridge
 import com.saveourtool.save.utils.orNotFound
 
 import org.springframework.data.repository.findByIdOrNull
 import org.springframework.transaction.annotation.Transactional
-import reactor.core.scheduler.Scheduler
-import reactor.core.scheduler.Schedulers
-
-import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.Dispatchers
 
 /**
  * Implementation of [S3KeyManager] which stores keys in database
  *
  * @param prefix a common prefix for all keys in S3 storage for this storage
  * @property repository repository for [E]
- * @property ioScheduler
- * @property ioDispatcher
+ * @property blockingBridge
  */
 abstract class AbstractS3KeyDatabaseManager<K : Any, E : BaseEntity, R : BaseEntityRepository<E>>(
     prefix: String,
     protected val repository: R,
-    val ioScheduler: Scheduler = Schedulers.boundedElastic(),
-    val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
+    val blockingBridge: BlockingBridge,
 ) : S3KeyManager<K> {
     /**
      * [S3KeyManager] with [Long] as key (it's [ID][com.saveourtool.save.spring.entity.BaseEntity.requiredId])

save-cloud-common/src/jvmMain/kotlin/com/saveourtool/save/storage/key/AbstractS3KeyDtoManager.kt

@@ -3,6 +3,7 @@ package com.saveourtool.save.storage.key
 import com.saveourtool.save.entities.DtoWithId
 import com.saveourtool.save.spring.entity.BaseEntityWithDtoWithId
 import com.saveourtool.save.spring.repository.BaseEntityRepository
+import com.saveourtool.save.utils.BlockingBridge
 import com.saveourtool.save.utils.orNotFound
 import org.springframework.data.repository.findByIdOrNull
 
@@ -11,11 +12,13 @@ import org.springframework.data.repository.findByIdOrNull
  *
  * @param prefix a common prefix for all keys in S3 storage for this storage
  * @param repository repository for [E] which is entity for [K]
+ * @param blockingBridge
  */
 abstract class AbstractS3KeyDtoManager<K : DtoWithId, E : BaseEntityWithDtoWithId<K>, R : BaseEntityRepository<E>>(
     prefix: String,
     repository: R,
-) : AbstractS3KeyDatabaseManager<K, E, R>(prefix, repository) {
+    blockingBridge: BlockingBridge,
+) : AbstractS3KeyDatabaseManager<K, E, R>(prefix, repository, blockingBridge) {
     override fun E.toKey(): K = toDto()
 
     override fun K.toEntity(): E = createNewEntityFromDto(this)

save-cloud-common/src/jvmMain/kotlin/com/saveourtool/save/storage/key/AbstractS3KeyEntityManager.kt

@@ -2,6 +2,7 @@ package com.saveourtool.save.storage.key
 
 import com.saveourtool.save.spring.entity.BaseEntity
 import com.saveourtool.save.spring.repository.BaseEntityRepository
+import com.saveourtool.save.utils.BlockingBridge
 import com.saveourtool.save.utils.orNotFound
 import org.springframework.data.repository.findByIdOrNull
 
@@ -10,13 +11,16 @@ import org.springframework.data.repository.findByIdOrNull
  *
  * @param prefix a common prefix for all keys in S3 storage for this storage
  * @param repository repository for [E]
+ * @param blockingBridge
  */
 abstract class AbstractS3KeyEntityManager<E : BaseEntity, R : BaseEntityRepository<E>>(
     prefix: String,
     repository: R,
+    blockingBridge: BlockingBridge,
 ) : AbstractS3KeyDatabaseManager<E, E, R>(
     prefix,
     repository,
+    blockingBridge,
 ) {
     override fun E.toKey(): E = this
 

save-cloud-common/src/jvmMain/kotlin/com/saveourtool/save/utils/BlockingBridge.kt

@@ -0,0 +1,58 @@
+package com.saveourtool.save.utils
+
+import org.jetbrains.annotations.NonBlocking
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
+import reactor.core.scheduler.Schedulers
+import reactor.kotlin.core.publisher.toMono
+
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+
+/**
+ * A bridge for blocking (IO) operations
+ *
+ * @property ioScheduler [Scheduler] for IO operations for [Mono] and [Flux]
+ * @property ioDispatcher [CoroutineDispatcher] for IO operations in suspend function
+ */
+class BlockingBridge(
+    val ioScheduler: Scheduler = Schedulers.boundedElastic(),
+    val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
+) {
+    /**
+     * Taking from https://projectreactor.io/docs/core/release/reference/#faq.wrap-blocking
+     *
+     * @param supplier blocking operation like JDBC
+     * @return [Mono] from result of blocking operation [T]
+     * @see blockingToFlux
+     */
+    @NonBlocking
+    fun <T : Any> blockingToMono(supplier: () -> T?): Mono<T> = supplier.toMono().subscribeOn(ioScheduler)
+
+    /**
+     * @param supplier blocking operation like JDBC
+     * @return [Flux] from result of blocking operation [List] of [T]
+     * @see blockingToMono
+     */
+    @NonBlocking
+    fun <T> blockingToFlux(supplier: () -> Iterable<T>): Flux<T> = blockingToMono(supplier).flatMapIterable { it }
+
+    /**
+     * @param supplier blocking operation like JDBC
+     * @return suspend result of blocking operation [T]
+     * @see blockingToMono
+     */
+    @NonBlocking
+    suspend fun <T> blockingToSuspend(supplier: () -> T): T = withContext(ioDispatcher) {
+        supplier()
+    }
+
+    companion object {
+        /**
+         * A default instance of [BlockingBridge]
+         */
+        val default: BlockingBridge = BlockingBridge()
+    }
+}