Add KRepository (with suspend functions)

api/src/main/kotlin/org/taymyr/play/repository/domain/KRepository.kt

@@ -0,0 +1,142 @@
+package org.taymyr.play.repository.domain
+
+import akka.Done
+
+/**
+ * DDD repository for identified aggregate (use coroutines).
+ */
+interface KRepository<Aggregate, Identity> {
+
+    /**
+     * Generate a new identifier.
+     */
+    fun nextIdentity(): Identity
+
+    /**
+     * Get aggregate by the identifier.
+     * @param id Identifier.
+     * @return aggregate or null if aggregate not exist.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun get(id: Identity): Aggregate?
+
+    /**
+     * Get all aggregates from the repository.
+     * @return List of aggregates or an empty list if the repository is empty.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun getAll(): List<Aggregate>
+
+    /**
+     * Finding aggregates on the repository by their identifiers.
+     * @param ids List of identifiers.
+     * @return List of aggregates or an empty list if aggregates not found.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun findByIds(ids: List<Identity>): List<Aggregate>
+
+    /**
+     * Finding aggregates on the repository by their identifiers.
+     * @param ids List of identifiers.
+     * @return List of aggregates or an empty list if aggregates not found.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun findByIds(vararg ids: Identity): List<Aggregate> = findByIds(ids.asList())
+
+    /**
+     * Removing aggregate from the repository.
+     * @param aggregate Aggregate.
+     * @return [Done] if removing successfully. Otherwise will throw an exception.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun remove(aggregate: Aggregate): Done
+
+    /**
+     * Removing aggregates from the repository.
+     * @param aggregates List of aggregates.
+     * @return [Done] if removing successfully. Otherwise will throw an exception.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun removeAll(aggregates: Collection<Aggregate>): Done
+
+    /**
+     * Create aggregate on the repository.
+     * @param aggregate Aggregate.
+     * @return [Done] if creation successfully. Otherwise will throw an exception.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun create(aggregate: Aggregate): Done
+
+    /**
+     * Create aggregates on the repository.
+     * @param aggregates Aggregates.
+     * @return [Done] if creation successfully. Otherwise will throw an exception.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun createAll(aggregates: Collection<Aggregate>): Done
+
+    /**
+     * Saving aggregate on the repository.
+     * @param aggregate Aggregate.
+     * @return [Done] if saving successfully. Otherwise will throw an exception.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun save(aggregate: Aggregate): Done
+
+    /**
+     * Saving aggregates on the repository.
+     * @param aggregates Aggregates.
+     * @return [Done] if saving successfully. Otherwise will throw an exception.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun saveAll(aggregates: Collection<Aggregate>): Done
+
+    /**
+     * Finding aggregates on the repository by jpaQuery.
+     * @param jpaQuery Jpa query.
+     * @param parameters Map of parameters name and value in jpa query.
+     * @return List of aggregates or an empty list if aggregates not found.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun findAggregates(jpaQuery: String, parameters: Map<String, Any>): List<Aggregate>
+
+    /**
+     * Finding aggregates on the repository by jpaQuery and using pagination.
+     * @param jpaQuery Jpa query.
+     * @param parameters Map of parameters name and value in jpa query.
+     * @param offset Offset from the beginning of the list
+     * @param limit The number of elements in the sample.
+     * @return List of aggregates or an empty list if aggregates not found.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun findAggregates(jpaQuery: String, parameters: Map<String, Any>, offset: Int, limit: Int): List<Aggregate>
+
+    /**
+     * Finding aggregate on the repository by jpaQuery and using pagination.
+     * @param jpaQuery Jpa query.
+     * @param parameters Map of parameters name and value in jpa query.
+     * @return aggregate or null if aggregate not exist.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun findAggregate(jpaQuery: String, parameters: Map<String, Any>): Aggregate?
+
+    /**
+     * Finding specific value on the repository by jpaQuery.
+     * @param jpaQuery Jpa query.
+     * @param parameters Map of parameters name and value in jpa query.
+     * @param clazz Class of the find value.
+     * @return List of aggregates or an empty list if aggregates not found.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun <E> findValues(jpaQuery: String, parameters: Map<String, Any>, clazz: Class<E>): List<E>
+
+    /**
+     * Find specific value on the repository by jpaQuery.
+     * @param jpaQuery Jpa query.
+     * @param parameters Map of parameters name and value in jpa query.
+     * @param clazz Class of the find value.
+     * @return Specific value or null.
+     * @throws Exception Any exceptions while execute a query on the database will wrapped.
+     */
+    suspend fun <E> findValue(jpaQuery: String, parameters: Map<String, Any>, clazz: Class<E>): E?
+}

buildSrc/src/main/kotlin/Versions.kt

@@ -1,6 +1,7 @@
 @Suppress("ObjectPropertyName")
 object Versions {
     const val kotlin = "1.5.20"
+    const val kotlinCoroutines = "1.5.0"
     const val scalaBinary = "2.13"
     const val lagom = "1.6.4" // "1.5.5"
     const val play = "2.8.2"

jpa/build.gradle.kts

@@ -21,6 +21,7 @@ compileTestKotlin.kotlinOptions.freeCompilerArgs += listOf("-Xjvm-default=all",
 
 dependencies {
     implementation(kotlin("stdlib-jdk8"))
+    implementation("org.jetbrains.kotlinx", "kotlinx-coroutines-core", Versions.kotlinCoroutines)
     api(project(":play-repository-api-java"))
     compileOnly("com.typesafe.play", "play-java-jpa_$scalaBinaryVersion", playVersion)
     implementation("org.hibernate", "hibernate-entitymanager", Versions.hibernateVersion)

jpa/src/main/kotlin/org/taymyr/play/repository/infrastructure/persistence/JPABaseRepository.kt

@@ -0,0 +1,16 @@
+package org.taymyr.play.repository.infrastructure.persistence
+
+import play.db.jpa.JPAApi
+import java.io.Serializable
+import javax.persistence.EntityManager
+
+abstract class JPABaseRepository<Aggregate : Any, Identity : Serializable> @JvmOverloads constructor(
+    protected val jpaApi: JPAApi,
+    protected val executionContext: DatabaseExecutionContext,
+    protected val clazz: Class<out Aggregate>,
+    protected val persistenceUnitName: String = "default"
+) {
+    protected fun <E> transaction(function: (EntityManager) -> E): E = jpaApi.withTransaction(persistenceUnitName, function)
+
+    protected fun <E> readOnly(function: (EntityManager) -> E): E = jpaApi.withTransaction(persistenceUnitName, true, function)
+}

jpa/src/main/kotlin/org/taymyr/play/repository/infrastructure/persistence/JPARepository.kt

@@ -9,28 +9,23 @@ import java.util.Optional
 import java.util.Optional.ofNullable
 import java.util.concurrent.CompletableFuture.supplyAsync
 import java.util.concurrent.CompletionStage
-import java.util.function.Supplier
 import javax.persistence.EntityManager
 
 /**
  * JPA implementation of DDD repository for aggregates.
  */
 abstract class JPARepository<Aggregate : Any, Identity : Serializable> @JvmOverloads constructor(
-    protected val jpaApi: JPAApi,
-    protected val executionContext: DatabaseExecutionContext,
-    protected val clazz: Class<out Aggregate>,
-    protected val persistenceUnitName: String = "default"
-) : Repository<Aggregate, Identity> {
-
-    protected fun <E> transaction(function: (EntityManager) -> E): E = jpaApi.withTransaction(persistenceUnitName, function)
-
-    protected fun <E> readOnly(function: (EntityManager) -> E): E = jpaApi.withTransaction(persistenceUnitName, true, function)
+    jpaApi: JPAApi,
+    executionContext: DatabaseExecutionContext,
+    clazz: Class<out Aggregate>,
+    persistenceUnitName: String = "default"
+) : JPABaseRepository<Aggregate, Identity>(jpaApi, executionContext, clazz, persistenceUnitName), Repository<Aggregate, Identity> {
 
     protected fun <E> execute(function: (EntityManager) -> E): CompletionStage<E> =
-        supplyAsync(Supplier { transaction(function) }, executionContext)
+        supplyAsync({ transaction(function) }, executionContext)
 
     protected fun <E> executeRO(function: (EntityManager) -> E): CompletionStage<E> =
-        supplyAsync(Supplier { readOnly(function) }, executionContext)
+        supplyAsync({ readOnly(function) }, executionContext)
 
     protected fun <E> executeSession(function: (Session) -> E): CompletionStage<E> =
         execute { em -> function.invoke(em.unwrap(Session::class.java)) }

jpa/src/main/kotlin/org/taymyr/play/repository/infrastructure/persistence/KJPARepository.kt

@@ -0,0 +1,117 @@
+package org.taymyr.play.repository.infrastructure.persistence
+
+import akka.Done
+import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.withContext
+import org.hibernate.Session
+import org.taymyr.play.repository.domain.KRepository
+import play.db.jpa.JPAApi
+import java.io.Serializable
+import java.lang.IllegalArgumentException
+import javax.persistence.EntityManager
+
+abstract class KJPARepository <Aggregate : Any, Identity : Serializable> @JvmOverloads constructor(
+    jpaApi: JPAApi,
+    executionContext: DatabaseExecutionContext,
+    clazz: Class<out Aggregate>,
+    persistenceUnitName: String = "default"
+) : JPABaseRepository<Aggregate, Identity> (jpaApi, executionContext, clazz, persistenceUnitName), KRepository<Aggregate, Identity> {
+
+    private val dispatcher = executionContext.asCoroutineDispatcher()
+
+    private suspend fun <E> execute(function: (EntityManager) -> E): E = withContext(dispatcher) {
+        transaction { function.invoke(it) }
+    }
+
+    private suspend fun <E> executeRO(function: (EntityManager) -> E): E = withContext(dispatcher) {
+        readOnly { function.invoke(it) }
+    }
+
+    private suspend fun <E> executeSession(function: (Session) -> E): E = execute { em -> function.invoke(em.unwrap(Session::class.java)) }
+
+    private suspend fun <E> executeSessionRO(function: (Session) -> E): E = executeRO { em -> function.invoke(em.unwrap(Session::class.java)) }
+
+    override suspend fun get(id: Identity): Aggregate? = execute { em -> em.find(clazz, id) }
+
+    override suspend fun getAll(): List<Aggregate> = executeRO { em ->
+        val criteriaBuilder = em.criteriaBuilder
+        @Suppress("UNCHECKED_CAST")
+        val criteriaQuery = criteriaBuilder.createQuery(clazz as Class<Nothing>)
+        val root = criteriaQuery.from(clazz)
+        val all = criteriaQuery.select(root)
+        em.createQuery(all).resultList
+    }
+
+    override suspend fun findByIds(ids: List<Identity>): List<Aggregate> = executeSessionRO { session -> session.byMultipleIds(clazz).multiLoad(ids) }
+
+    override suspend fun remove(aggregate: Aggregate): Done = execute { em ->
+        if (em.contains(aggregate)) em.remove(aggregate)
+        else em.remove(em.merge(aggregate))
+        Done.getInstance()
+    }
+
+    override suspend fun removeAll(aggregates: Collection<Aggregate>): Done = execute { em ->
+        aggregates.forEach {
+            if (em.contains(it)) em.remove(it)
+            else em.remove(em.merge(it))
+        }
+        Done.getInstance()
+    }
+
+    override suspend fun create(aggregate: Aggregate): Done = execute { em ->
+        em.persist(aggregate)
+        Done.getInstance()
+    }
+
+    override suspend fun createAll(aggregates: Collection<Aggregate>): Done = execute { em ->
+        aggregates.forEach { em.persist(it) }
+        Done.getInstance()
+    }
+
+    override suspend fun save(aggregate: Aggregate): Done = execute { em ->
+        em.merge(aggregate)
+        Done.getInstance()
+    }
+
+    override suspend fun saveAll(aggregates: Collection<Aggregate>): Done = execute { em ->
+        aggregates.forEach { em.merge(it) }
+        Done.getInstance()
+    }
+
+    override suspend fun findAggregates(jpaQuery: String, parameters: Map<String, Any>): List<Aggregate> = executeRO { em ->
+        val query = em.createQuery(jpaQuery, clazz)
+        parameters.forEach { query.setParameter(it.key, it.value) }
+        query.resultList.toList()
+    }
+
+    override suspend fun findAggregates(jpaQuery: String, parameters: Map<String, Any>, offset: Int, limit: Int): List<Aggregate> {
+        if (offset < 0 || limit < 0) throw IllegalArgumentException("offset and limit must not be negative")
+        return executeRO { em ->
+            val query = em.createQuery(jpaQuery, clazz)
+            parameters.forEach { query.setParameter(it.key, it.value) }
+            query.firstResult = offset
+            query.maxResults = limit
+            query.resultList.toList()
+        }
+    }
+
+    override suspend fun findAggregate(jpaQuery: String, parameters: Map<String, Any>): Aggregate? = executeRO { em ->
+        val query = em.createQuery(jpaQuery, clazz)
+        parameters.forEach { query.setParameter(it.key, it.value) }
+        query.maxResults = 1
+        return@executeRO if (query.resultList.toList().isEmpty()) null else query.resultList[0]
+    }
+
+    override suspend fun <E> findValues(jpaQuery: String, parameters: Map<String, Any>, clazz: Class<E>): List<E> = executeRO { em ->
+        val query = em.createQuery(jpaQuery, clazz)
+        parameters.forEach { query.setParameter(it.key, it.value) }
+        query.resultList.toList()
+    }
+
+    override suspend fun <E> findValue(jpaQuery: String, parameters: Map<String, Any>, clazz: Class<E>): E? = executeRO { em ->
+        val query = em.createQuery(jpaQuery, clazz)
+        parameters.forEach { query.setParameter(it.key, it.value) }
+        query.maxResults = 1
+        return@executeRO if (query.resultList.toList().isEmpty()) null else query.resultList[0]
+    }
+}

jpa/src/test/kotlin/org/taymyr/play/repository/domain/UserRepository.kt

@@ -1,3 +1,5 @@
 package org.taymyr.play.repository.domain
 
 interface UserRepository : Repository<User, String>
+
+interface UserKRepository : KRepository<User, String>

jpa/src/test/kotlin/org/taymyr/play/repository/infrastructure/persistence/UserRepositoryImpl.kt

@@ -1,6 +1,7 @@
 package org.taymyr.play.repository.infrastructure.persistence
 
 import org.taymyr.play.repository.domain.User
+import org.taymyr.play.repository.domain.UserKRepository
 import org.taymyr.play.repository.domain.UserRepository
 import play.db.jpa.JPAApi
 import java.util.UUID
@@ -16,3 +17,11 @@ class UserRepositoryImpl @Inject constructor(
 
     override fun nextIdentity(): String = UUID.randomUUID().toString()
 }
+
+class UserKRepositoryImpl @Inject constructor(
+    jpaApi: JPAApi,
+    executionContext: DatabaseExecutionContext
+) : KJPARepository<User, String>(jpaApi, executionContext, UserImpl::class.java), UserKRepository {
+
+    override fun nextIdentity(): String = UUID.randomUUID().toString()
+}