Fix subscription resolver return type validation

src/main/kotlin/graphql/kickstart/tools/ResolverInfo.kt

@@ -68,7 +68,8 @@ internal class MultiResolverInfo(
 
 internal class RootResolverInfo(
     val resolvers: List<GraphQLRootResolver>,
-    private val options: SchemaParserOptions
+    private val options: SchemaParserOptions,
+    val isSubscription: Boolean = false
 ) : ResolverInfo() {
     override fun getFieldSearches() =
         resolvers.map { FieldResolverScanner.Search(getRealResolverClass(it, options), this, it) }

src/main/kotlin/graphql/kickstart/tools/SchemaClassScanner.kt

@@ -543,7 +543,7 @@ internal class SchemaClassScanner(
 
         private val queryResolverInfo = RootResolverInfo(queryResolvers, options)
         private val mutationResolverInfo = RootResolverInfo(mutationResolvers, options)
-        private val subscriptionResolverInfo = RootResolverInfo(subscriptionResolvers, options)
+        private val subscriptionResolverInfo = RootResolverInfo(subscriptionResolvers, options, isSubscription = true)
 
         val query = createRootType("query", queryDefinition, queryName, true, queryResolvers, GraphQLQueryResolver::class.java, queryResolverInfo)
         val mutation = createRootType("mutation", mutationDefinition, mutationName, rootInfo.isMutationRequired(), mutationResolvers, GraphQLMutationResolver::class.java, mutationResolverInfo)

src/main/kotlin/graphql/kickstart/tools/resolver/FieldResolverScanner.kt

@@ -2,7 +2,6 @@ package graphql.kickstart.tools.resolver
 
 import graphql.GraphQLContext
 import graphql.Scalars
-import graphql.kickstart.tools.GraphQLSubscriptionResolver
 import graphql.kickstart.tools.ResolverInfo
 import graphql.kickstart.tools.RootResolverInfo
 import graphql.kickstart.tools.SchemaParserOptions
@@ -13,10 +12,10 @@ import graphql.schema.DataFetchingEnvironment
 import kotlinx.coroutines.channels.ReceiveChannel
 import org.apache.commons.lang3.ClassUtils
 import org.apache.commons.lang3.reflect.FieldUtils
+import org.apache.commons.lang3.reflect.TypeUtils
 import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
 import java.lang.reflect.*
-import java.util.concurrent.CompletableFuture
 import kotlin.reflect.full.valueParameters
 import kotlin.reflect.jvm.javaType
 import kotlin.reflect.jvm.kotlinFunction
@@ -124,27 +123,27 @@ internal class FieldResolverScanner(val options: SchemaParserOptions) {
             // to avoid issues with duplicate method declarations
             .filter { it.declaringClass != Object::class.java }
             // subscription resolvers must return a publisher
-            .filter { search.source !is GraphQLSubscriptionResolver || resolverMethodReturnsPublisher(it) }
+            .filter { !search.isSubscription || resolverMethodReturnsPublisher(it) }
             .toList()
     }
 
     private fun resolverMethodReturnsPublisher(method: Method) =
-        method.returnType.isAssignableFrom(Publisher::class.java)
+        // suspend functions and unbounded type variables are erased to Object, so the actual return type is unknown here
+        method.returnType == Any::class.java
+            || Publisher::class.java.isAssignableFrom(method.returnType)
             || resolverMethodReturnsPublisherFuture(method)
             || receiveChannelToPublisherWrapper(method)
 
     private fun resolverMethodReturnsPublisherFuture(method: Method) =
-        method.returnType.isAssignableFrom(CompletableFuture::class.java)
-            && method.genericReturnType is ParameterizedType
-            && (method.genericReturnType as ParameterizedType).actualTypeArguments
-            .any {
-                it is ParameterizedType && it.unwrap().isAssignableFrom(Publisher::class.java)
-            }
+        method.genericReturnType.futureValueType()
+            ?.let { if (it is WildcardType) it.upperBounds.first() else it }
+            ?.let { TypeUtils.isAssignable(it, Publisher::class.java) } == true
 
     private fun receiveChannelToPublisherWrapper(method: Method) =
-        method.returnType.isAssignableFrom(ReceiveChannel::class.java)
+        ReceiveChannel::class.java.isAssignableFrom(method.returnType)
             && options.genericWrappers.any { wrapper ->
             val isReceiveChannelWrapper = wrapper.type == method.returnType
+            // lambdas are erased to return Object, so this only checks that the transformer can return a Publisher
             val hasPublisherTransformer = wrapper
                 .transformer.javaClass
                 .declaredMethods
@@ -199,7 +198,7 @@ internal class FieldResolverScanner(val options: SchemaParserOptions) {
 
         searches.forEach { search ->
             signatures.addAll(getMissingMethodSignatures(field, search, isBoolean, scannedProperties))
-            isSubscription = isSubscription || search.source is GraphQLSubscriptionResolver
+            isSubscription = isSubscription || search.isSubscription
         }
 
         val sourceName = field.sourceLocation?.sourceName ?: "<unknown>"
@@ -241,7 +240,9 @@ internal class FieldResolverScanner(val options: SchemaParserOptions) {
         val resolverInfo: ResolverInfo,
         val source: Any?,
         val requiredFirstParameterType: Class<*>? = null
-    )
+    ) {
+        val isSubscription get() = resolverInfo is RootResolverInfo && resolverInfo.isSubscription
+    }
 }
 
 internal class FieldResolverError(msg: String) : RuntimeException(msg)

src/main/kotlin/graphql/kickstart/tools/resolver/MethodFieldResolver.kt

@@ -6,6 +6,8 @@
 import graphql.kickstart.tools.SchemaParserOptions.GenericWrapper
 import graphql.kickstart.tools.util.JavaType
 import graphql.kickstart.tools.util.coroutineScope
+import graphql.kickstart.tools.util.futureValueType
+import graphql.kickstart.tools.util.typeArgument
 import graphql.kickstart.tools.util.unwrap
 import graphql.language.*
 import graphql.schema.DataFetcher
@@ -14,6 +16,8 @@
 import graphql.schema.GraphQLTypeUtil.isScalar
 import graphql.schema.LightDataFetcher
 import kotlinx.coroutines.future.future
+import org.apache.commons.lang3.reflect.TypeUtils
+import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
 import java.lang.reflect.InvocationTargetException
 import java.lang.reflect.Method
@@ -109,7 +113,7 @@
                                 "please insert it into the GraphQLContext map when building the ExecutionInput. " +
                                 "This warning will become an error in the future."
                         )
                         environment.getContext() // TODO: remove deprecated use in next major release
                     }
                 }
 
@@ -141,7 +145,7 @@
             is ListType -> List::class.java.isAssignableFrom(this.genericType.getRawClass(genericParameterType))
                 && isConcreteScalarType(environment, type.type, this.genericType.unwrapGenericType(genericParameterType))
 
             is TypeName -> environment.graphQLSchema?.getType(type.name!!)?.let { isScalar(it) && type.name != "ID" }
                 ?: false
 
             is NonNullType -> isConcreteScalarType(environment, type.type, genericParameterType)
@@ -150,20 +154,26 @@
     }
 
     override fun scanForMatches(): List<TypeClassMatcher.PotentialMatch> {
-        val unwrappedGenericType = genericType.unwrapGenericType(
-            try {
-                method.kotlinFunction?.returnType?.javaType ?: method.genericReturnType
-            } catch (e: InternalError) {
-                method.genericReturnType
-            }
-        )
+        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))
 
         return field.inputValueDefinitions.mapIndexed { i, inputDefinition ->
             TypeClassMatcher.PotentialMatch.parameterType(inputDefinition.type, getMethodParameterType(i)!!, genericType, SchemaClassScanner.MethodParameterReference(method, i))
         } + listOf(returnValueMatch)
     }
 
+    // generic wrappers only match their exact type, so a Publisher implementation (e.g. Flux), or a future of one,
+    // is matched as a plain Publisher of its events
+    private fun JavaType.asPublisher(): JavaType =
+        (futureValueType() ?: this).typeArgument(Publisher::class.java)
+            ?.let { TypeUtils.parameterize(Publisher::class.java, it) }
+            ?: this
+
     private fun getIndexOffset(): Int {
         return if (resolverInfo is DataClassTypeResolverInfo && !method.declaringClass.isAssignableFrom(resolverInfo.dataClassType)) {
             1

src/main/kotlin/graphql/kickstart/tools/util/Utils.kt

@@ -6,9 +6,12 @@
 import graphql.schema.DataFetchingEnvironment
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import org.apache.commons.lang3.reflect.TypeUtils
 import java.lang.reflect.Method
 import java.lang.reflect.ParameterizedType
 import java.lang.reflect.Proxy
+import java.util.concurrent.CompletionStage
+import java.util.concurrent.Future
 
 /**
  * @author Andrew Potter
@@ -37,8 +40,14 @@
         this as Class<*>
     }
 
+internal fun JavaType.typeArgument(type: Class<*>): JavaType? =
+    TypeUtils.getTypeArguments(this, type)?.get(type.typeParameters.first())
+
+internal fun JavaType.futureValueType(): JavaType? =
+    typeArgument(CompletionStage::class.java) ?: typeArgument(Future::class.java)
+
 internal fun DataFetchingEnvironment.coroutineScope(): CoroutineScope {
     val context: Any? = this.getContext()
     return if (context is CoroutineScope) context else CoroutineScope(Dispatchers.Default)
 }