Support for suspendable property resolvers [#38]

src/main/kotlin/com/github/pgutkowski/kgraphql/schema/dsl/PropertyDSL.kt

@@ -40,6 +40,24 @@ class PropertyDSL<T : Any, R>(val name : String, block : PropertyDSL<T, R>.() ->
     fun <E, W, Q, A, S>resolver(function: (T, E, W, Q, A, S) -> R)
             = resolver(FunctionWrapper.on(function, true))
 
+    fun suspendResolver(function: suspend (T) -> R)
+            = resolver(FunctionWrapper.onSuspend(function, true))
+
+    fun <E>suspendResolver(function: suspend (T, E) -> R)
+            = resolver(FunctionWrapper.onSuspend(function, true))
+
+    fun <E, W>suspendResolver(function: suspend (T, E, W) -> R)
+            = resolver(FunctionWrapper.onSuspend(function, true))
+
+    fun <E, W, Q>suspendResolver(function: suspend (T, E, W, Q) -> R)
+            = resolver(FunctionWrapper.onSuspend(function, true))
+
+    fun <E, W, Q, A>suspendResolver(function: suspend (T, E, W, Q, A) -> R)
+            = resolver(FunctionWrapper.onSuspend(function, true))
+
+    fun <E, W, Q, A, S>suspendResolver(function: suspend (T, E, W, Q, A, S) -> R)
+            = resolver(FunctionWrapper.onSuspend(function, true))
+
     fun accessRule(rule: (T, Context) -> Exception?){
 
         val accessRuleAdapter: (T?, Context) -> Exception? = { parent, ctx ->

src/main/kotlin/com/github/pgutkowski/kgraphql/schema/execution/ParallelRequestExecutor.kt

@@ -18,11 +18,8 @@ import com.github.pgutkowski.kgraphql.schema.scalar.serializeScalar
 import com.github.pgutkowski.kgraphql.schema.structure2.Field
 import com.github.pgutkowski.kgraphql.schema.structure2.InputValue
 import com.github.pgutkowski.kgraphql.schema.structure2.Type
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Job
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 import kotlin.coroutines.CoroutineContext
 import kotlin.reflect.KProperty1
 
@@ -51,36 +48,14 @@ class ParallelRequestExecutor(val schema: DefaultSchema) : RequestExecutor, Coro
     override suspend fun suspendExecute(plan: ExecutionPlan, variables: VariablesJson, context: Context): String {
         val root = jsonNodeFactory.objectNode()
         val data = root.putObject("data")
-        val channel = Channel<Pair<Execution, JsonNode>>()
-        val jobs = plan
-                .map { execution ->
-                    launch(dispatcher) {
-                        try {
-                            val writeOperation = writeOperation(
-                                    ctx = ExecutionContext(Variables(schema, variables, execution.variables), context),
-                                    node = execution,
-                                    operation = execution.field as Field.Function<*, *>
-                            )
-                            channel.send(execution to writeOperation)
-                        } catch (e: Exception) {
-                            channel.close(e)
-                        }
-                    }
-                }
-                .toList()
 
-        //intermediate data structure necessary to preserve ordering
-        val resultMap = mutableMapOf<Execution, JsonNode>()
-        repeat(plan.size) {
-            try {
-                val (execution, jsonNode) = channel.receive()
-                resultMap.put(execution, jsonNode)
-            } catch (e: Exception) {
-                jobs.forEach { it.cancel() }
-                throw e
-            }
+        val resultMap = plan.toMapAsync() {
+            writeOperation(
+                ctx = ExecutionContext(Variables(schema, variables, it.variables), context),
+                node = it,
+                operation = it.field as Field.Function<*, *>
+            )
         }
-        channel.close()
 
         for (operation in plan) {
             data.set(operation.aliasOrKey, resultMap[operation])
@@ -93,6 +68,33 @@ class ParallelRequestExecutor(val schema: DefaultSchema) : RequestExecutor, Coro
         suspendExecute(plan, variables, context)
     }
 
+    private suspend fun <T, R> Collection<T>.toMapAsync(block: suspend (T) -> R): Map<T, R> = coroutineScope {
+        val channel = Channel<Pair<T, R>>()
+        val jobs = map { item ->
+            launch(dispatcher) {
+                try {
+                    val res = block(item)
+                    channel.send(item to res)
+                } catch (e: Exception) {
+                    channel.close(e)
+                }
+            }
+        }
+        val resultMap = mutableMapOf<T, R>()
+        repeat(size) {
+            try {
+                val (item, result) = channel.receive()
+                resultMap[item] = result
+            } catch (e: Exception) {
+                jobs.forEach(Job::cancel)
+                throw e
+            }
+        }
+
+        channel.close()
+        resultMap
+    }
+
     private suspend fun <T> writeOperation(ctx: ExecutionContext, node: Execution.Node, operation: FunctionWrapper<T>): JsonNode {
         node.field.checkAccess(null, ctx.requestContext)
         val operationResult: T? = operation.invoke(
@@ -134,9 +136,12 @@ class ParallelRequestExecutor(val schema: DefaultSchema) : RequestExecutor, Coro
             //check value, not returnType, because this method can be invoked with element value
             value is Collection<*> -> {
                 if (returnType.isList()) {
-                    val arrayNode = jsonNodeFactory.arrayNode(value.size)
-                    value.forEach { element -> arrayNode.add(createNode(ctx, element, node, returnType.unwrapList())) }
-                    arrayNode
+                    val valuesMap = value.toMapAsync {
+                        createNode(ctx, it, node, returnType.unwrapList())
+                    }
+                    value.fold(jsonNodeFactory.arrayNode(value.size)) { array, v ->
+                        array.add(valuesMap[v])
+                    }
                 } else {
                     throw ExecutionException("Invalid collection value for non collection property")
                 }

src/test/kotlin/com/github/pgutkowski/kgraphql/TestUtils.kt

@@ -25,15 +25,15 @@ fun getMap(map : Map<*,*>, key : String) : Map<*,*>{
 fun <T> Map<*, *>.extract(path: String) : T {
     val tokens = path.trim().split('/').filter(String::isNotBlank)
     try {
-        return tokens.fold(this as Any?, { workingMap, token ->
+        return tokens.fold(this as Any?) { workingMap, token ->
             if(token.contains('[')){
                 val list = (workingMap as Map<*,*>)[token.substringBefore('[')]
-                val index = token[token.indexOf('[')+1].toString().toInt()
+                val index = token.substring(token.indexOf('[')+1, token.length -1).toInt()
                 (list as List<*>)[index]
             } else {
                 (workingMap as Map<*,*>)[token]
             }
-        }) as T
+        } as T
     } catch (e : Exception){
         throw IllegalArgumentException("Path: $path does not exist in map: ${this}", e)
     }
@@ -61,8 +61,8 @@ fun assertError(map : Map<*,*>, vararg messageElements : String) {
     MatcherAssert.assertThat(errorMessage, CoreMatchers.notNullValue())
 
     messageElements
-            .filterNot { errorMessage.contains(it) }
-            .forEach { throw AssertionError("Expected error message to contain $it, but was: $errorMessage") }
+        .filterNot { errorMessage.contains(it) }
+        .forEach { throw AssertionError("Expected error message to contain $it, but was: $errorMessage") }
 }
 
 inline fun <reified T: Exception> expect(message: String? = null, block: () -> Unit){

src/test/kotlin/com/github/pgutkowski/kgraphql/access/AccessRulesTest.kt

@@ -1,10 +1,6 @@
 package com.github.pgutkowski.kgraphql.access
 
-import com.github.pgutkowski.kgraphql.context
-import com.github.pgutkowski.kgraphql.defaultSchema
-import com.github.pgutkowski.kgraphql.deserialize
-import com.github.pgutkowski.kgraphql.expect
-import com.github.pgutkowski.kgraphql.extract
+import com.github.pgutkowski.kgraphql.*
 import org.hamcrest.CoreMatchers.equalTo
 import org.hamcrest.MatcherAssert.assertThat
 import org.junit.Test
@@ -24,10 +20,16 @@ class AccessRulesTest {
         }
 
         type<Player>{
+            val accessRuleBlock = { player: Player, _: Context ->
+                if (player.name != "BONNER") IllegalAccessException("ILLEGAL ACCESS") else null
+            }
+
             property(Player::id){
-                accessRule { player, _ ->
-                    if(player.name != "BONNER") IllegalAccessException("ILLEGAL ACCESS") else null
-                }
+                accessRule(accessRuleBlock)
+            }
+            property<String>("item") {
+                accessRule(accessRuleBlock)
+                resolver { "item" }
             }
         }
     }
@@ -44,13 +46,28 @@ class AccessRulesTest {
 
     @Test
     fun `reject when not matching`(){
-        expect<IllegalAccessException> {
+        expect<IllegalAccessException>("") {
             deserialize (
                     schema.execute("{ black_mamba {id} }", context { +"LAKERS" })
             ).extract<String>("data/black_mamba/id")
         }
     }
 
+    @Test
+    fun `allow property resolver access rule`() {
+        assertThat(
+            deserialize(schema.execute("{white_mamba {item}}")).extract<String>("data/white_mamba/item"),
+            equalTo("item")
+        )
+    }
+
+    @Test
+    fun `reject property resolver access rule`() {
+        expect<IllegalAccessException>("ILLEGAL ACCESS") {
+            schema.execute("{black_mamba {item}}", context { +"LAKERS" })
+        }
+    }
+
     //TODO: MORE TESTS
 
 }

src/test/kotlin/com/github/pgutkowski/kgraphql/integration/ParallelExecutionTest.kt

@@ -1,39 +1,71 @@
 package com.github.pgutkowski.kgraphql.integration
 
 import com.github.pgutkowski.kgraphql.KGraphQL
-import com.github.pgutkowski.kgraphql.assertNoErrors
 import com.github.pgutkowski.kgraphql.extract
 import com.github.pgutkowski.kgraphql.deserialize
 import kotlinx.coroutines.delay
 import org.hamcrest.CoreMatchers
 import org.hamcrest.MatcherAssert
 import org.junit.Test
+import kotlin.random.Random
 
 class ParallelExecutionTest {
 
+    data class AType(val id: Int)
+
     val syncResolversSchema = KGraphQL.schema {
         repeat(1000) {
-            query("automated-${it}") {
+            query("automated-$it") {
                 resolver { ->
                     Thread.sleep(3)
-                    "${it}"
+                    "$it"
                 }
             }
         }
     }
 
     val suspendResolverSchema = KGraphQL.schema {
         repeat(1000) {
-            query("automated-${it}") {
+            query("automated-$it") {
                 suspendResolver { ->
                     delay(3)
-                    "${it}"
+                    "$it"
+                }
+            }
+        }
+    }
+
+    val suspendPropertySchema = KGraphQL.schema {
+        query("getAll") {
+            resolver { -> (0..999).map { AType(it) } }
+        }
+        type<AType> {
+            property<List<AType>>("children") {
+                suspendResolver { parent ->
+                    (0..50).map {
+                        delay(Random.nextLong(1, 100))
+                        AType((parent.id * 10) + it)
+                    }
                 }
             }
         }
     }
 
-    val query = "{ " + (0..999).map { "automated-${it}" }.joinToString(", ") + " }"
+    @Test
+    fun `Suspendable property resolvers`() {
+        val query = "{getAll{id,children{id}}}"
+        val map = deserialize(suspendPropertySchema.execute(query))
+
+        MatcherAssert.assertThat(map.extract<Int>("data/getAll[0]/id"), CoreMatchers.equalTo(0))
+        MatcherAssert.assertThat(map.extract<Int>("data/getAll[500]/id"), CoreMatchers.equalTo(500))
+        MatcherAssert.assertThat(map.extract<Int>("data/getAll[766]/id"), CoreMatchers.equalTo(766))
+
+        MatcherAssert.assertThat(map.extract<Int>("data/getAll[5]/children[5]/id"), CoreMatchers.equalTo(55))
+        MatcherAssert.assertThat(map.extract<Int>("data/getAll[75]/children[9]/id"), CoreMatchers.equalTo(759))
+        MatcherAssert.assertThat(map.extract<Int>("data/getAll[888]/children[50]/id"), CoreMatchers.equalTo(8930))
+    }
+
+    val query = "{ " + (0..999).map { "automated-$it" }.joinToString(", ") + " }"
 
     @Test
     fun `1000 synchronous resolvers sleeping with Thread sleep`(){

src/test/kotlin/com/github/pgutkowski/kgraphql/specification/language/ArgumentsSpecificationTest.kt

@@ -30,6 +30,30 @@ class ArgumentsSpecificationTest {
                     }.take(size)
                 }
             }
+            property<Int>("none") {
+                suspendResolver { actor -> actor.age }
+            }
+            property<Int>("one") {
+                suspendResolver {actor, one: Int -> actor.age + one }
+            }
+            property<Int>("two") {
+                suspendResolver { actor, one: Int, two: Int -> actor.age + one + two }
+            }
+            property<Int>("three") {
+                suspendResolver { actor, one: Int, two: Int, three: Int ->
+                    actor.age + one + two + three
+                }
+            }
+            property<Int>("four") {
+                suspendResolver { actor, one: Int, two: Int, three: Int, four: Int ->
+                    actor.age + one + two + three + four
+                }
+            }
+            property<Int>("five") {
+                suspendResolver { actor, one: Int, two: Int, three: Int, four: Int, five: Int ->
+                    actor.age + one + two + three + four + five
+                }
+            }
         }
     }
 
@@ -50,5 +74,31 @@ class ArgumentsSpecificationTest {
         )
     }
 
+    @Test
+    fun `all arguments to suspendResolvers`() {
+        val request = """
+            {
+                actor {
+                    none
+                    one(one: 1)
+                    two(one: 2, two: 3)
+                    three(one: 4, two: 5, three: 6)
+                    four(one: 7, two: 8, three: 9, four: 10)
+                    five(one: 11, two: 12, three: 13, four: 14, five: 15)
+                }
+            }
+        """.trimIndent()
+        val response = deserialize(schema.execute(request)) as Map<String, Any>
+        assertThat(response, equalTo(mapOf<String, Any>(
+                "data" to mapOf("actor" to mapOf(
+                        "none" to age,
+                        "one" to age + 1,
+                        "two" to age + 2 + 3,
+                        "three" to age + 4 + 5 + 6,
+                        "four" to age + 7 + 8 + 9 + 10,
+                        "five" to age + 11 + 12 + 13 + 14 + 15
+                ))
+        )))
+    }
 
 }