feat: Add subscriptions support to ktor server

examples/server/ktor-server/build.gradle.kts

@@ -14,6 +14,8 @@ application {
 dependencies {
     implementation("com.expediagroup", "graphql-kotlin-ktor-server")
     implementation(libs.ktor.server.netty)
+    implementation(libs.ktor.server.websockets)
+    implementation(libs.ktor.server.cors)
     implementation(libs.logback)
     implementation(libs.kotlinx.coroutines.jdk8)
 }

examples/server/ktor-server/src/main/kotlin/com/expediagroup/graphql/examples/server/ktor/GraphQLModule.kt

@@ -18,6 +18,7 @@ package com.expediagroup.graphql.examples.server.ktor
 import com.expediagroup.graphql.dataloader.KotlinDataLoaderRegistryFactory
 import com.expediagroup.graphql.examples.server.ktor.schema.BookQueryService
 import com.expediagroup.graphql.examples.server.ktor.schema.CourseQueryService
+import com.expediagroup.graphql.examples.server.ktor.schema.ExampleSubscriptionService
 import com.expediagroup.graphql.examples.server.ktor.schema.HelloQueryService
 import com.expediagroup.graphql.examples.server.ktor.schema.LoginMutationService
 import com.expediagroup.graphql.examples.server.ktor.schema.UniversityQueryService
@@ -28,12 +29,25 @@ import com.expediagroup.graphql.server.ktor.GraphQL
 import com.expediagroup.graphql.server.ktor.graphQLGetRoute
 import com.expediagroup.graphql.server.ktor.graphQLPostRoute
 import com.expediagroup.graphql.server.ktor.graphQLSDLRoute
+import com.expediagroup.graphql.server.ktor.graphQLSubscriptionsRoute
 import com.expediagroup.graphql.server.ktor.graphiQLRoute
+import io.ktor.serialization.jackson.JacksonWebsocketContentConverter
 import io.ktor.server.application.Application
 import io.ktor.server.application.install
+import io.ktor.server.plugins.cors.routing.CORS
 import io.ktor.server.routing.Routing
+import io.ktor.server.websocket.WebSockets
+import io.ktor.server.websocket.pingPeriod
+import java.time.Duration
 
 fun Application.graphQLModule() {
+    install(WebSockets) {
+        pingPeriod = Duration.ofSeconds(1)
+        contentConverter = JacksonWebsocketContentConverter()
+    }
+    install(CORS) {
+        anyHost()
+    }
     install(GraphQL) {
         schema {
             packages = listOf("com.expediagroup.graphql.examples.server")
@@ -46,6 +60,9 @@ fun Application.graphQLModule() {
             mutations = listOf(
                 LoginMutationService()
             )
+            subscriptions = listOf(
+                ExampleSubscriptionService()
+            )
         }
         engine {
             dataLoaderRegistryFactory = KotlinDataLoaderRegistryFactory(
@@ -59,6 +76,7 @@ fun Application.graphQLModule() {
     install(Routing) {
         graphQLGetRoute()
         graphQLPostRoute()
+        graphQLSubscriptionsRoute()
         graphiQLRoute()
         graphQLSDLRoute()
     }

examples/server/ktor-server/src/main/kotlin/com/expediagroup/graphql/examples/server/ktor/schema/ExampleSubscriptionService.kt

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2023 Expedia, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.expediagroup.graphql.examples.server.ktor.schema
+
+import com.expediagroup.graphql.generator.annotations.GraphQLDescription
+import com.expediagroup.graphql.server.operations.Subscription
+import graphql.GraphqlErrorException
+import graphql.execution.DataFetcherResult
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.reactive.asPublisher
+import org.reactivestreams.Publisher
+import kotlin.random.Random
+
+class ExampleSubscriptionService : Subscription {
+
+    @GraphQLDescription("Returns a single value")
+    fun singleValue(): Flow<Int> = flowOf(1)
+
+    @GraphQLDescription("Returns stream of values")
+    fun multipleValues(): Flow<Int> = flowOf(1, 2, 3)
+
+    @GraphQLDescription("Returns a random number every second")
+    suspend fun counter(limit: Int? = null): Flow<Int> = flow {
+        var count = 0
+        while (true) {
+            count++
+            if (limit != null) {
+                if (count > limit) break
+            }
+            emit(Random.nextInt())
+            delay(1000)
+        }
+    }
+
+    @GraphQLDescription("Returns a random number every second, errors if even")
+    fun counterWithError(): Flow<Int> = flow {
+        while (true) {
+            val value = Random.nextInt()
+            if (value % 2 == 0) {
+                throw Exception("Value is even $value")
+            } else emit(value)
+            delay(1000)
+        }
+    }
+
+    @GraphQLDescription("Returns one value then an error")
+    fun singleValueThenError(): Flow<Int> = flowOf(1, 2)
+        .map { if (it == 2) throw Exception("Second value") else it }
+
+    @GraphQLDescription("Returns stream of errors")
+    fun flowOfErrors(): Publisher<DataFetcherResult<String?>> {
+        val dfr: DataFetcherResult<String?> = DataFetcherResult.newResult<String?>()
+            .data(null)
+            .error(GraphqlErrorException.newErrorException().cause(Exception("error thrown")).build())
+            .build()
+
+        return flowOf(dfr, dfr).asPublisher()
+    }
+}

examples/server/ktor-server/src/main/resources/logback.xml

@@ -4,7 +4,7 @@
             <pattern>%d{YYYY-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
         </encoder>
     </appender>
-    <root level="trace">
+    <root level="debug">
         <appender-ref ref="STDOUT"/>
     </root>
     <logger name="org.eclipse.jetty" level="INFO"/>

gradle/libs.versions.toml

@@ -68,9 +68,12 @@ ktor-client-apache = { group = "io.ktor", name = "ktor-client-apache", version.r
 ktor-client-cio = { group = "io.ktor", name = "ktor-client-cio", version.ref = "ktor" }
 ktor-client-content = { group = "io.ktor", name = "ktor-client-content-negotiation", version.ref = "ktor" }
 ktor-client-serialization = { group = "io.ktor", name = "ktor-client-serialization", version.ref = "ktor" }
+ktor-client-websockets = { group = "io.ktor", name = "ktor-client-websockets", version.ref = "ktor" }
 ktor-serialization-jackson = { group = "io.ktor", name = "ktor-serialization-jackson", version.ref = "ktor" }
 ktor-server-core = { group = "io.ktor", name = "ktor-server-core", version.ref = "ktor" }
 ktor-server-content = { group = "io.ktor", name = "ktor-server-content-negotiation", version.ref = "ktor" }
+ktor-server-websockets = { group = "io.ktor", name = "ktor-server-websockets", version.ref = "ktor" }
+ktor-server-cors = { group = "io.ktor", name = "ktor-server-cors", version.ref = "ktor" }
 maven-plugin-annotations = { group = "org.apache.maven.plugin-tools", name = "maven-plugin-annotations", version.ref = "maven-plugin-annotation" }
 maven-plugin-api = { group = "org.apache.maven", name = "maven-plugin-api", version.ref = "maven-plugin-api" }
 maven-project = { group = "org.apache.maven", name = "maven-project", version.ref = "maven-project" }

servers/graphql-kotlin-ktor-server/build.gradle.kts

@@ -10,8 +10,10 @@ dependencies {
     api(libs.ktor.serialization.jackson)
     api(libs.ktor.server.core)
     api(libs.ktor.server.content)
+    api(libs.ktor.server.websockets)
     testImplementation(libs.kotlinx.coroutines.test)
     testImplementation(libs.ktor.client.content)
+    testImplementation(libs.ktor.client.websockets)
     testImplementation(libs.ktor.server.cio)
     testImplementation(libs.ktor.server.test.host)
 }

servers/graphql-kotlin-ktor-server/src/main/kotlin/com/expediagroup/graphql/server/ktor/GraphQL.kt

@@ -32,7 +32,12 @@ import com.expediagroup.graphql.generator.federation.FederatedSchemaGeneratorHoo
 import com.expediagroup.graphql.generator.federation.FederatedSimpleTypeResolver
 import com.expediagroup.graphql.generator.federation.toFederatedSchema
 import com.expediagroup.graphql.generator.internal.state.ClassScanner
+import com.expediagroup.graphql.server.execution.DefaultGraphQLSubscriptionExecutor
 import com.expediagroup.graphql.server.execution.GraphQLRequestHandler
+import com.expediagroup.graphql.server.ktor.subscriptions.KtorGraphQLSubscriptionHandler
+import com.expediagroup.graphql.server.ktor.subscriptions.DefaultKtorGraphQLSubscriptionHooks
+import com.expediagroup.graphql.server.ktor.subscriptions.graphqlws.KtorGraphQLWebSocketProtocolHandler
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import graphql.execution.AsyncExecutionStrategy
 import graphql.execution.AsyncSerialExecutionStrategy
 import graphql.execution.instrumentation.ChainedInstrumentation
@@ -90,7 +95,7 @@ class GraphQL(config: GraphQLConfiguration) {
             config = schemaConfig,
             queries = config.schema.queries.toTopLevelObjects(),
             mutations = config.schema.mutations.toTopLevelObjects(),
-            subscriptions = emptyList(),
+            subscriptions = config.schema.subscriptions.toTopLevelObjects(),
             schemaObject = config.schema.schemaObject?.let { TopLevelObject(it) }
         )
     } else {
@@ -107,7 +112,7 @@ class GraphQL(config: GraphQLConfiguration) {
             gen.generateSchema(
                 queries = config.schema.queries.toTopLevelObjects(),
                 mutations = config.schema.mutations.toTopLevelObjects(),
-                subscriptions = emptyList(),
+                subscriptions = config.schema.subscriptions.toTopLevelObjects(),
                 schemaObject = config.schema.schemaObject?.let { TopLevelObject(it) }
             )
         }
@@ -160,6 +165,17 @@ class GraphQL(config: GraphQLConfiguration) {
         )
     )
 
+    val subscriptionsHandler: KtorGraphQLSubscriptionHandler by lazy {
+        KtorGraphQLWebSocketProtocolHandler(
+            subscriptionExecutor = DefaultGraphQLSubscriptionExecutor(
+                graphQL = engine,
+                dataLoaderRegistryFactory = config.engine.dataLoaderRegistryFactory,
+            ),
+            objectMapper = jacksonObjectMapper().apply(config.server.jacksonConfiguration),
+            subscriptionHooks = DefaultKtorGraphQLSubscriptionHooks(),
+        )
+    }
+
     companion object Plugin : BaseApplicationPlugin<Application, GraphQLConfiguration, GraphQL> {
         override val key: AttributeKey<GraphQL> = AttributeKey("GraphQL")
 

servers/graphql-kotlin-ktor-server/src/main/kotlin/com/expediagroup/graphql/server/ktor/GraphQLConfiguration.kt

@@ -22,12 +22,13 @@ import com.expediagroup.graphql.dataloader.KotlinDataLoaderRegistryFactory
 import com.expediagroup.graphql.generator.TopLevelNames
 import com.expediagroup.graphql.generator.execution.KotlinDataFetcherFactoryProvider
 import com.expediagroup.graphql.generator.execution.SimpleKotlinDataFetcherFactoryProvider
-import com.expediagroup.graphql.generator.hooks.NoopSchemaGeneratorHooks
+import com.expediagroup.graphql.generator.hooks.FlowSubscriptionSchemaGeneratorHooks
 import com.expediagroup.graphql.generator.hooks.SchemaGeneratorHooks
 import com.expediagroup.graphql.generator.scalars.IDValueUnboxer
 import com.expediagroup.graphql.server.Schema
 import com.expediagroup.graphql.server.operations.Mutation
 import com.expediagroup.graphql.server.operations.Query
+import com.expediagroup.graphql.server.operations.Subscription
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import graphql.execution.DataFetcherExceptionHandler
@@ -116,15 +117,14 @@ class GraphQLConfiguration(config: ApplicationConfig) {
         var queries: List<Query> = emptyList()
         /** List of GraphQL mutations supported by this server */
         var mutations: List<Mutation> = emptyList()
-        // TODO support subscriptions
-//        /** List of GraphQL subscriptions supported by this server */
-//        var subscriptions: List<Subscription> = emptyList()
+        /** List of GraphQL subscriptions supported by this server */
+        var subscriptions: List<Subscription> = emptyList()
         /** GraphQL schema object with any custom directives */
         var schemaObject: Schema? = null
         /** The names of the top level objects in the schema, defaults to Query, Mutation and Subscription */
         var topLevelNames: TopLevelNames = TopLevelNames()
         /** Custom hooks that will be used when generating the schema */
-        var hooks: SchemaGeneratorHooks = NoopSchemaGeneratorHooks
+        var hooks: SchemaGeneratorHooks = FlowSubscriptionSchemaGeneratorHooks()
         /** Apollo Federation configuration */
         val federation: FederationConfiguration = FederationConfiguration(config)
         fun federation(federationConfig: FederationConfiguration.() -> Unit) {

servers/graphql-kotlin-ktor-server/src/main/kotlin/com/expediagroup/graphql/server/ktor/GraphQLRoutes.kt

@@ -17,6 +17,7 @@
 package com.expediagroup.graphql.server.ktor
 
 import com.expediagroup.graphql.generator.extensions.print
+import com.expediagroup.graphql.server.ktor.subscriptions.KtorGraphQLSubscriptionHandler
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.ktor.http.ContentType
 import io.ktor.serialization.jackson.jackson
@@ -29,6 +30,7 @@ import io.ktor.server.routing.Route
 import io.ktor.server.routing.application
 import io.ktor.server.routing.get
 import io.ktor.server.routing.post
+import io.ktor.server.websocket.webSocket
 
 /**
  * Configures GraphQL GET route
@@ -70,6 +72,27 @@ fun Route.graphQLPostRoute(endpoint: String = "graphql", streamingResponse: Bool
     return route
 }
 
+/**
+ * Configures GraphQL subscriptions route
+ *
+ * @param endpoint GraphQL server subscriptions endpoint, defaults to 'subscriptions'
+ * @param handlerOverride Alternative KtorGraphQLSubscriptionHandler to handle subscriptions logic
+ */
+fun Route.graphQLSubscriptionsRoute(
+    endpoint: String = "subscriptions",
+    protocol: String? = null,
+    handlerOverride: KtorGraphQLSubscriptionHandler? = null,
+) {
+    val handler = handlerOverride ?: run {
+        val graphQLPlugin = this.application.plugin(GraphQL)
+        graphQLPlugin.subscriptionsHandler
+    }
+
+    webSocket(path = endpoint, protocol = protocol) {
+        handler.handle(this)
+    }
+}
+
 /**
  * Configures GraphQL SDL route.
  *
@@ -88,14 +111,18 @@ fun Route.graphQLSDLRoute(endpoint: String = "sdl"): Route {
  *
  * @param endpoint GET endpoint that will return instance of GraphiQL IDE, defaults to 'graphiql'
  * @param graphQLEndpoint your GraphQL endpoint for processing requests
+ * @param subscriptionsEndpoint your GraphQL subscriptions endpoint
  */
-fun Route.graphiQLRoute(endpoint: String = "graphiql", graphQLEndpoint: String = "graphql"): Route {
+fun Route.graphiQLRoute(
+    endpoint: String = "graphiql",
+    graphQLEndpoint: String = "graphql",
+    subscriptionsEndpoint: String = "subscriptions",
+): Route {
     val contextPath = this.environment?.rootPath
     val graphiQL = GraphQL::class.java.classLoader.getResourceAsStream("graphql-graphiql.html")?.bufferedReader()?.use { reader ->
         reader.readText()
             .replace("\${graphQLEndpoint}", if (contextPath.isNullOrBlank()) graphQLEndpoint else "$contextPath/$graphQLEndpoint")
-            .replace("\${subscriptionsEndpoint}", if (contextPath.isNullOrBlank()) "subscriptions" else "$contextPath/subscriptions")
-//            .replace("\${subscriptionsEndpoint}", if (contextPath.isBlank()) config.routing.subscriptions.endpoint else "$contextPath/${config.routing.subscriptions.endpoint}")
+            .replace("\${subscriptionsEndpoint}", if (contextPath.isNullOrBlank()) subscriptionsEndpoint else "$contextPath/$subscriptionsEndpoint")
     } ?: throw IllegalStateException("Unable to load GraphiQL")
     return get(endpoint) {
         call.respondText(graphiQL, ContentType.Text.Html)

servers/graphql-kotlin-ktor-server/src/main/kotlin/com/expediagroup/graphql/server/ktor/KtorGraphQLServer.kt

@@ -26,5 +26,5 @@ import io.ktor.server.request.ApplicationRequest
 class KtorGraphQLServer(
     requestParser: KtorGraphQLRequestParser,
     contextFactory: KtorGraphQLContextFactory,
-    requestHandler: GraphQLRequestHandler
+    requestHandler: GraphQLRequestHandler,
 ) : GraphQLServer<ApplicationRequest>(requestParser, contextFactory, requestHandler)

servers/graphql-kotlin-ktor-server/src/main/kotlin/com/expediagroup/graphql/server/ktor/subscriptions/KtorGraphQLSubscriptionHandler.kt

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2023 Expedia, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.expediagroup.graphql.server.ktor.subscriptions
+
+import io.ktor.server.websocket.WebSocketServerSession
+
+interface KtorGraphQLSubscriptionHandler {
+    suspend fun handle(session: WebSocketServerSession)
+}