How to correctly enable native subscription support for Kotlin Flow in graphql-kotlin-spring-server?

[antonywhatever]

Based on the content of https://github.com/ExpediaGroup/graphql-kotlin/blob/master/graphql-kotlin-schema-generator/src/test/kotlin/com/expediagroup/graphql/execution/FlowSubscriptionExecutionStrategyTest.kt I was expecting to enable support with the two simple steps below but an attempt to transform GraphQL bean results in BeanCurrentlyInCreationException. I have tried various workarounds e.g. @DependsOn("graphQL") or changing a name for new bean.

@Configuration
class Configuration {

    @Bean
    fun maxQueryDepthInstrumentation(): MaxQueryDepthInstrumentation = MaxQueryDepthInstrumentation(13)

    @Bean
    fun maxQueryComplexityInstrumentation(): MaxQueryComplexityInstrumentation =
        MaxQueryComplexityInstrumentation(maxQueryComplexity) { ... }

    // 1st step
    @Bean
    fun schemaGeneratorHooks(): SchemaGeneratorHooks = object : FlowSubscriptionSchemaGeneratorHooks() {
        override fun willGenerateGraphQLType(type: KType): GraphQLType? = when (type.classifier) {
            LocalDate::class -> ExtendedScalars.Date
            YetOneCustom::class -> yetOneCustomGraphQLScalarType
            else -> null
        }
    }

    // 2nd step
    @Bean
    fun graphQL(@Autowired graphQL: GraphQL): GraphQL = graphQL.transform { it.subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy()) }
}

Then I tried to change 2nd step for

    // 2nd step
    @Primary
    @Bean
    fun graphQL(graphQLSchema: GraphQLSchema): GraphQL = GraphQL.newGraphQL(graphQLSchema)
            .subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy())
            .build()

Also I tried to set spring.main.allow-bean-definition-overriding=true and override Spring bean

    @Component
    class BeansInitializer : ApplicationContextInitializer<GenericApplicationContext> {
    
        override fun initialize(applicationContext: GenericApplicationContext) {
            val graphQL = (applicationContext.getBean("graphQL") as GraphQL).transform {
                it.subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy())
            }
            applicationContext.registerBean("graphQL", graphql.GraphQL::class.java, graphQL)
        }
    }

    @Component
    class Runner(private val applicationContext: GenericApplicationContext) : ApplicationRunner {
    
        override fun run(args: ApplicationArguments?) {
            val graphQL = (applicationContext.getBean("graphQL") as GraphQL).transform {
                it.subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy())
            }
            applicationContext.registerBean("graphQL", graphql.GraphQL::class.java, graphQL)
        }
    }

[dariuszkuc]

Hello 👋
Thanks for raising this. It appears that while graphql-kotlin-schema-generator exposes FlowSubscriptionExecutionStrategy we forgot to update graphql-kotlin-spring-server to easily use it.

Currently you will have to recreate the GraphQL bean (https://github.com/ExpediaGroup/graphql-kotlin/blob/master/graphql-kotlin-spring-server/src/main/kotlin/com/expediagroup/graphql/spring/GraphQLSchemaConfiguration.kt#L61-L94) and configure it with the flow strategy.

I'll open up a PR to update the default to FlowSubscriptionExecutionStrategy.

[antonywhatever]

FYI If you make https://github.com/ExpediaGroup/graphql-kotlin/blob/master/examples/server/spring-server/src/main/kotlin/com/expediagroup/graphql/examples/hooks/CustomSchemaGeneratorHooks.kt extend FlowSubscriptionSchemaGeneratorHooks and put that GraphQL bean into https://github.com/ExpediaGroup/graphql-kotlin/blob/master/examples/server/spring-server/src/main/kotlin/com/expediagroup/graphql/examples/Application.kt and configure it with the flow strategy, the subscriptions will stop working with

{
  "error": "Could not connect to websocket endpoint ws://localhost:8080/subscriptions. Please check if the endpoint url is correct."
}

Changing the strategy back to SubscriptionExecutionStrategy will fix this.

[dariuszkuc]

Indeed, the underlying issue is that all our subscription handling logic was based on the reactive-streams Publisher API so when you apply existing Flow subscription strategy it would cause class cast exception. Instead of rewriting existing subscription handling logic I updated flow subscription strategy to also return a publisher -> #972