Handle exception on subscription initialization (#1940)

### 📝 Description

In case of Subscription initiation failure, for example exception thrown
after some custom authentication/authorisation mechanism throws
exception, error was not passed through as response body, because of
exception:
```
java.lang.NullPointerException: graphQL.execute(input)
    …<Flow<ExecutionResult>>() must not be null
   at com.expediagroup.graphql.server.execution.GraphQLRequestHandler.executeSubscription(GraphQLRequestHandler.kt:167)
 ```

### 🔗 Related Issues
-

Author: pdambrauskas

servers/graphql-kotlin-server/src/main/kotlin/com/expediagroup/graphql/server/execution/GraphQLRequestHandler.kt

@@ -44,6 +44,7 @@ import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.future.await
 import kotlinx.coroutines.supervisorScope
@@ -61,6 +62,7 @@ open class GraphQLRequestHandler(
                         .firstOrNull(Instrumentation::isBatchDataLoaderInstrumentation)
                         ?.javaClass
                 }
+
                 instrumentation.isBatchDataLoaderInstrumentation() -> instrumentation.javaClass
                 else -> null
             }
@@ -81,6 +83,7 @@ open class GraphQLRequestHandler(
                 val batchGraphQLContext = graphQLContext + getBatchContext(1, dataLoaderRegistry)
                 execute(graphQLRequest, batchGraphQLContext, dataLoaderRegistry)
             }
+
             is GraphQLBatchRequest -> {
                 if (graphQLRequest.containsMutation()) {
                     val batchGraphQLContext = graphQLContext + getBatchContext(1, dataLoaderRegistry)
@@ -145,9 +148,11 @@ open class GraphQLRequestHandler(
             DataLoaderLevelDispatchedInstrumentation::class.java -> mapOf(
                 ExecutionLevelDispatchedState::class to ExecutionLevelDispatchedState(batchSize)
             )
+
             DataLoaderSyncExecutionExhaustedInstrumentation::class.java -> mapOf(
                 SyncExecutionExhaustedState::class to SyncExecutionExhaustedState(batchSize, dataLoaderRegistry)
             )
+
             else -> emptyMap<Any, Any>()
         }
         return batchContext
@@ -162,9 +167,12 @@ open class GraphQLRequestHandler(
     ): Flow<GraphQLResponse<*>> {
         val dataLoaderRegistry = dataLoaderRegistryFactory?.generate(graphQLContext)
         val input = graphQLRequest.toExecutionInput(graphQLContext, dataLoaderRegistry)
+        val executionResult = graphQL.execute(input)
+
+        val resultFlow: Flow<ExecutionResult> = executionResult
+            .getData<Flow<ExecutionResult>?>() ?: flowOf(executionResult)
 
-        return graphQL.execute(input)
-            .getData<Flow<ExecutionResult>>()
+        return resultFlow
             .map { result -> result.toGraphQLResponse() }
             .catch { throwable ->
                 val error = throwable.toGraphQLError()

servers/graphql-kotlin-server/src/test/kotlin/com/expediagroup/graphql/server/execution/GraphQLRequestHandlerTest.kt

@@ -22,12 +22,15 @@ import com.expediagroup.graphql.dataloader.instrumentation.level.DataLoaderLevel
 import com.expediagroup.graphql.dataloader.instrumentation.syncexhaustion.DataLoaderSyncExecutionExhaustedInstrumentation
 import com.expediagroup.graphql.generator.SchemaGeneratorConfig
 import com.expediagroup.graphql.generator.TopLevelObject
+import com.expediagroup.graphql.generator.execution.FlowSubscriptionExecutionStrategy
 import com.expediagroup.graphql.generator.extensions.toGraphQLContext
+import com.expediagroup.graphql.generator.hooks.FlowSubscriptionSchemaGeneratorHooks
 import com.expediagroup.graphql.generator.toSchema
 import com.expediagroup.graphql.server.types.GraphQLBatchRequest
 import com.expediagroup.graphql.server.types.GraphQLBatchResponse
 import com.expediagroup.graphql.server.types.GraphQLRequest
 import com.expediagroup.graphql.server.types.GraphQLResponse
+import com.expediagroup.graphql.server.types.GraphQLServerError
 import graphql.ExecutionInput
 import graphql.GraphQL
 import graphql.GraphQLContext
@@ -38,24 +41,32 @@ import graphql.schema.DataFetchingEnvironment
 import graphql.schema.GraphQLSchema
 import io.mockk.every
 import io.mockk.mockk
-import kotlinx.coroutines.runBlocking
-import org.dataloader.DataLoader
-import org.dataloader.DataLoaderFactory
-import org.junit.jupiter.api.Test
 import java.util.concurrent.CompletableFuture
 import kotlin.random.Random
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.runBlocking
+import org.dataloader.DataLoader
+import org.dataloader.DataLoaderFactory
+import org.junit.jupiter.api.Test
 
 class GraphQLRequestHandlerTest {
 
     private val testSchema: GraphQLSchema = toSchema(
-        config = SchemaGeneratorConfig(supportedPackages = listOf("com.expediagroup.graphql.server.execution")),
+        config = SchemaGeneratorConfig(
+            supportedPackages = listOf("com.expediagroup.graphql.server.execution"),
+            hooks = FlowSubscriptionSchemaGeneratorHooks(),
+        ),
         queries = listOf(TopLevelObject(BasicQuery())),
-        mutations = listOf(TopLevelObject(BasicMutation()))
+        mutations = listOf(TopLevelObject(BasicMutation())),
+        subscriptions = listOf(TopLevelObject(BasicSubscription())),
     )
-    private val testGraphQL: GraphQL = GraphQL.newGraphQL(testSchema).build()
+    private val testGraphQL: GraphQL = GraphQL.newGraphQL(testSchema).subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy()).build()
     private val graphQLRequestHandler = GraphQLRequestHandler(testGraphQL)
 
     private fun getBatchingRequestHandler(instrumentation: Instrumentation): GraphQLRequestHandler =
@@ -317,6 +328,59 @@ class GraphQLRequestHandlerTest {
         }
     }
 
+    @Test
+    fun `execute graphQL subscription`() {
+        val response = runBlocking {
+            val context = emptyMap<String, Any>().toGraphQLContext()
+            val request = GraphQLRequest(
+                query = "subscription { users(name: \"Jane Doe\") { name } }",
+            )
+            graphQLRequestHandler.executeSubscription(request, context).first()
+        }
+
+        assertNotNull(response.data as? Map<*, *>) { data ->
+            assertNotNull(data["users"] as? Map<*, *>) { user ->
+                assertEquals("Jane Doe", user["name"])
+            }
+        }
+        assertNull(response.errors)
+        assertNull(response.extensions)
+    }
+
+    @Test
+    fun `execute graphQL subscription with error`() {
+        val response = runBlocking {
+            val context = emptyMap<String, Any>().toGraphQLContext()
+            val request = GraphQLRequest(
+                query = "subscription { withFlowError { name } }",
+            )
+            graphQLRequestHandler.executeSubscription(request, context).first()
+        }
+
+        assertNotNull(response.errors as List<GraphQLServerError>) { errors ->
+            assertNotNull(errors[0]) { error ->
+                assertEquals("Subscription failure", error.message)
+            }
+        }
+    }
+
+    @Test
+    fun `execute graphQL subscription with error on init`() {
+        val response = runBlocking {
+            val context = emptyMap<String, Any>().toGraphQLContext()
+            val request = GraphQLRequest(
+                query = "subscription { withInitError { name } }",
+            )
+            graphQLRequestHandler.executeSubscription(request, context).first()
+        }
+
+        assertNotNull(response.errors as List<GraphQLServerError>) { errors ->
+            assertNotNull(errors[0]) { error ->
+                assertEquals("Exception while fetching data (/withInitError) : Subscription failure", error.message)
+            }
+        }
+    }
+
     data class User(val id: Int, val name: String)
 
     class BasicQuery {
@@ -338,4 +402,16 @@ class GraphQLRequestHandlerTest {
     class BasicMutation {
         fun addUser(name: String): User = User(Random.nextInt(), name)
     }
+
+    class BasicSubscription {
+        fun users(name: String): Flow<User> = flowOf(User(Random.nextInt(), name))
+
+        fun withFlowError(): Flow<User> = flow {
+            throw Exception("Subscription failure")
+        }
+
+        fun withInitError(): Flow<User> {
+            throw Exception("Subscription failure")
+        }
+    }
 }