Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ import org.slf4j.LoggerFactory
import java.lang.reflect.InvocationTargetException
import java.lang.reflect.Method
import java.util.*
import java.util.concurrent.CompletionStage
import java.util.concurrent.Future
import java.util.function.Supplier
import kotlin.coroutines.intrinsics.suspendCoroutineUninterceptedOrReturn
import kotlin.reflect.jvm.javaType
Expand Down Expand Up @@ -126,12 +128,17 @@ internal class MethodFieldResolver(
}
}

// the class the return type unwraps to, so a value that happens to implement Future isn't mistaken for a wrapper.
// Object (e.g. for a union) can't tell them apart, so then every future of a method declared to return Future is checked
val valueClass = (TypeUtils.getRawType(getUnwrappedReturnType(), null) ?: Any::class.java)
.takeUnless { it == Any::class.java && TypeUtils.getRawType(getReturnType(), null) == Future::class.java }

return if (numberOfParameters > 0 || isSuspendFunction) {
// requires arguments and environment or is a suspend function
MethodFieldResolverDataFetcher(createSourceResolver(), this.method, args, options, isSuspendFunction)
MethodFieldResolverDataFetcher(createSourceResolver(), this.method, args, options, isSuspendFunction, valueClass)
} else {
// if there are no parameters an optimized version of the data fetcher can be used
LightMethodFieldResolverDataFetcher(createSourceResolver(), this.method, options)
LightMethodFieldResolverDataFetcher(createSourceResolver(), this.method, options, valueClass)
}
}

Expand All @@ -158,13 +165,7 @@ internal class MethodFieldResolver(
}

override fun scanForMatches(): List<TypeClassMatcher.PotentialMatch> {
val returnType = try {
method.kotlinFunction?.returnType?.javaType ?: method.genericReturnType
} catch (e: InternalError) {
method.genericReturnType
}
val unwrappedGenericType = genericType.unwrapGenericType(if (search.isSubscription) returnType.asPublisher() else returnType)
val returnValueMatch = TypeClassMatcher.PotentialMatch.returnValue(field.type, unwrappedGenericType, genericType, SchemaClassScanner.ReturnValueReference(method))
val returnValueMatch = TypeClassMatcher.PotentialMatch.returnValue(field.type, getUnwrappedReturnType(), genericType, SchemaClassScanner.ReturnValueReference(method))

return field.inputValueDefinitions.mapIndexed { i, inputDefinition ->
TypeClassMatcher.PotentialMatch.parameterType(inputDefinition.type, getMethodParameterType(i)!!, genericType, SchemaClassScanner.MethodParameterReference(method, i))
Expand All @@ -178,6 +179,16 @@ internal class MethodFieldResolver(
?.let { TypeUtils.parameterize(Publisher::class.java, it) }
?: this

private fun getReturnType(): JavaType =
try {
method.kotlinFunction?.returnType?.javaType ?: method.genericReturnType
} catch (e: InternalError) {
method.genericReturnType
}

private fun getUnwrappedReturnType(): JavaType =
getReturnType().let { genericType.unwrapGenericType(if (search.isSubscription) it.asPublisher() else it) }

private fun getIndexOffset(): Int {
return if (resolverInfo is DataClassTypeResolverInfo && !method.declaringClass.isAssignableFrom(resolverInfo.dataClassType)) {
1
Expand Down Expand Up @@ -207,7 +218,8 @@ internal class MethodFieldResolverDataFetcher(
private val method: Method,
private val args: List<ArgumentPlaceholder>,
private val options: SchemaParserOptions,
private val isSuspendFunction: Boolean
private val isSuspendFunction: Boolean,
private val valueClass: Class<*>?
) : DataFetcher<Any> {

override fun get(environment: DataFetchingEnvironment): Any? {
Expand All @@ -220,9 +232,11 @@ internal class MethodFieldResolverDataFetcher(
environment.coroutineScope().future(options.coroutineContextProvider.provide(), CoroutineStart.UNDISPATCHED) {
ensureActive()
invokeSuspend(source, method, args)?.transformWithGenericWrapper(options.genericWrappers) { environment }
.requireAwaitable(valueClass, method, isSuspendFunction = true)
}
} else {
invoke(method, source, args)?.transformWithGenericWrapper(options.genericWrappers) { environment }
.requireAwaitable(valueClass, method, isSuspendFunction = false)
}
}

Expand All @@ -241,13 +255,15 @@ internal class MethodFieldResolverDataFetcher(
internal class LightMethodFieldResolverDataFetcher(
private val sourceResolver: SourceResolver,
private val method: Method,
private val options: SchemaParserOptions
private val options: SchemaParserOptions,
private val valueClass: Class<*>?
) : LightDataFetcher<Any?> {

override fun get(fieldDefinition: GraphQLFieldDefinition?, sourceObject: Any?, environmentSupplier: Supplier<DataFetchingEnvironment>): Any? {
val source = sourceResolver.resolve(null, sourceObject)

return invoke(method, source, emptyArray())?.transformWithGenericWrapper(options.genericWrappers, environmentSupplier)
.requireAwaitable(valueClass, method, isSuspendFunction = false)
}

override fun get(environment: DataFetchingEnvironment): Any? {
Expand Down Expand Up @@ -278,6 +294,27 @@ private fun Any.transformWithGenericWrapper(
return genericWrapper.transformer(this, environmentSupplier.get())
}

// graphql-java only waits on a CompletionStage returned by a data fetcher, so any other future, or a future
// that a suspend function completes with, would reach it as the field value
private fun Any?.requireAwaitable(valueClass: Class<*>?, method: Method, isSuspendFunction: Boolean): Any? {
if (this !is Future<*> || valueClass?.isInstance(this) == true) {
return this
}

val resolverName = "${method.declaringClass.name}.${method.name}"
if (isSuspendFunction) {
throw ResolverError("Suspend function '$resolverName' resolved to a ${javaClass.name}, which graphql-java can't wait on. Return its value instead.")
}
if (this !is CompletionStage<*>) {
throw ResolverError(
"Resolver method '$resolverName' resolved to a ${javaClass.name}, which graphql-java can't wait on. " +
"Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead."
)
}

return this
}

private class CompareGenericWrappers {
companion object : Comparator<GenericWrapper> {
override fun compare(w1: GenericWrapper, w2: GenericWrapper): Int = when {
Expand Down
206 changes: 206 additions & 0 deletions src/test/kotlin/graphql/kickstart/tools/ReactiveTest.kt
Original file line number Diff line number Diff line change
@@ -1,14 +1,18 @@
package graphql.kickstart.tools

import graphql.ExceptionWhileDataFetching
import graphql.ExecutionResult
import graphql.GraphQL
import graphql.execution.AsyncExecutionStrategy
import graphql.kickstart.tools.SchemaParser.Companion.newParser
import graphql.kickstart.tools.SchemaParserOptions.Companion.newOptions
import graphql.kickstart.tools.SchemaParserOptions.GenericWrapper
import kotlinx.coroutines.ExperimentalCoroutinesApi
import org.junit.Test
import java.util.*
import java.util.concurrent.CompletableFuture
import java.util.concurrent.Future
import java.util.concurrent.FutureTask

//import io.reactivex.Single;
//import io.reactivex.internal.operators.single.SingleJust;
Expand Down Expand Up @@ -39,6 +43,208 @@ class ReactiveTest {
}
}

@Test
fun `future that is not a completion stage fails with a clear error`() {
val gql = GraphQL.newGraphQL(itemSchema(PlainFutureQuery())).build()

val result = gql.execute(
"""
query {
item { id }
itemWithArgument(id: 1) { id }
name
}
""")

assertEquals(result.getData(), mapOf("item" to null, "itemWithArgument" to null, "name" to null))
assertEquals(result.errorMessagesByPath(), mapOf(
"item" to "Resolver method '${PlainFutureQuery::class.java.name}.item' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead.",
"itemWithArgument" to "Resolver method '${PlainFutureQuery::class.java.name}.itemWithArgument' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead.",
"name" to "Resolver method '${PlainFutureQuery::class.java.name}.name' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead."
))
}

@Test
fun `suspend function that resolves to a future fails with a clear error`() {
val gql = GraphQL.newGraphQL(itemSchema(PlainFutureQuery())).build()

val result = gql.execute(
"""
query {
suspendItem { id }
suspendCompletableItem { id }
}
""")

assertEquals(result.getData(), mapOf("suspendItem" to null, "suspendCompletableItem" to null))
assertEquals(result.errorMessagesByPath(), mapOf(
"suspendItem" to "Suspend function '${PlainFutureQuery::class.java.name}.suspendItem' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return its value instead.",
"suspendCompletableItem" to "Suspend function '${PlainFutureQuery::class.java.name}.suspendCompletableItem' resolved to a java.util.concurrent.CompletableFuture, " +
"which graphql-java can't wait on. Return its value instead."
))
}

@Test
fun `future that is not a completion stage can be unwrapped by a generic wrapper`() {
val options = newOptions()
.genericWrappers(GenericWrapper.withTransformer(FutureTask::class, 0, { task -> task.get() }))
.build()
val gql = GraphQL.newGraphQL(itemSchema(PlainFutureQuery(), options)).build()

val data = assertNoGraphQlErrors(gql) {
"""
query {
item { id }
itemWithArgument(id: 2) { id }
name
suspendItem { id }
nullItem { id }
}
"""
}

assertEquals(data, mapOf(
"item" to mapOf("id" to "1"),
"itemWithArgument" to mapOf("id" to "2"),
"name" to "name",
"suspendItem" to mapOf("id" to "3"),
"nullItem" to null
))
}

@Test
fun `future that a generic wrapper, a type variable or a union resolves to fails with a clear error`() {
val schema = newParser()
.schemaString(
"""
type Query {
item: Item
boxedItem: Item
search: SearchResult
}

union SearchResult = Item

type Item {
id: ID
}
""")
.resolvers(FutureItemQuery())
.options(newOptions()
.genericWrappers(GenericWrapper.withTransformer(Box::class, 0, { box -> FutureTask { box.value }.also { it.run() } }))
.build())
.build()
.makeExecutableSchema()
val gql = GraphQL.newGraphQL(schema).build()

val result = gql.execute(
"""
query {
item { id }
boxedItem { id }
search {
... on Item { id }
}
}
""")

assertEquals(result.getData(), mapOf("item" to null, "boxedItem" to null, "search" to null))
assertEquals(result.errorMessagesByPath(), mapOf(
"item" to "Resolver method '${ItemQuery::class.java.name}.item' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead.",
"boxedItem" to "Resolver method '${FutureItemQuery::class.java.name}.boxedItem' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead.",
"search" to "Resolver method '${FutureItemQuery::class.java.name}.search' resolved to a java.util.concurrent.FutureTask, " +
"which graphql-java can't wait on. Return a CompletionStage, such as a CompletableFuture, from the method or its generic wrapper transformer instead."
))
}

@Test
fun `value that implements Future resolves normally`() {
val schema = newParser()
.schemaString(
"""
type Query {
job: Job
}

type Job {
id: ID
}
""")
.resolvers(object : GraphQLQueryResolver {
fun job(): Job = Job(1)
})
.build()
.makeExecutableSchema()

val gql = GraphQL.newGraphQL(schema).build()

val data = assertNoGraphQlErrors(gql) {
"""
query {
job { id }
}
"""
}

assertEquals(data, mapOf("job" to mapOf("id" to "1")))
}

private fun itemSchema(query: GraphQLQueryResolver, options: SchemaParserOptions = newOptions().build()) = newParser()
.schemaString(
"""
type Query {
item: Item
itemWithArgument(id: ID): Item
name: String
suspendItem: Item
suspendCompletableItem: Item
nullItem: Item
}

type Item {
id: ID
}
""")
.resolvers(query)
.options(options)
.build()
.makeExecutableSchema()

private fun ExecutionResult.errorMessagesByPath() =
errors.associate { it.path?.joinToString(".") to (it as ExceptionWhileDataFetching).exception.message }

// the futures are already done, so even a future that could be read without blocking is rejected
private class PlainFutureQuery : GraphQLQueryResolver {
fun item(): Future<Item> = FutureTask { Item(1) }.also { it.run() }
fun itemWithArgument(id: Long): Future<Item> = FutureTask { Item(id) }.also { it.run() }
fun name(): Future<String> = FutureTask { "name" }.also { it.run() }
suspend fun suspendItem(): Future<Item> = FutureTask { Item(3) }.also { it.run() }
suspend fun suspendCompletableItem(): Future<Item> = CompletableFuture.completedFuture(Item(4))
fun nullItem(): Future<Item?> = FutureTask<Item?> { null }.also { it.run() }
}

open class ItemQuery<R>(private val item: () -> R) {
fun item(): R = item.invoke()
}

private class FutureItemQuery : ItemQuery<Future<Item>>({ FutureTask { Item(1) }.also { it.run() } }), GraphQLQueryResolver {
fun boxedItem(): Box<Item> = Box(Item(2))
fun search(): Future<Any> = FutureTask<Any> { Item(3) }.also { it.run() }
}

private class Box<T>(val value: T)

private class Item(private val id: Long)

// implements Future, but is the schema type itself rather than a wrapper around one
private class Job(private val id: Long) : Future<String> by CompletableFuture.completedFuture("done")

private class Query : GraphQLQueryResolver {
// Single<Optional<Organization>> organization(int organizationid) {
// return Single.just(Optional.empty()); //CompletableFuture.completedFuture(null);
Expand Down
Loading