Kotlin Flow subscriptions don't return any values

[milgner]

Looking at the network connection, it seems like flows are always immediately treated as complete.
I have set up SchemaGeneratorConfig with hooks = FlowSubscriptionSchemaGeneratorHooks() and the GraphQL instance using subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy()).

There aren't any errors but my subscription only ever immediately sends one next message with an empty data payload, followed by the complete message, terminating the subscription.

I assume I'm doing something wrong when setting up the flow. I'm trying to forward Vert.x event bus messages to the subscribed clients:

import io.vertx.core.Vertx
import io.vertx.kotlin.coroutines.toReceiveChannel
import kotlinx.coroutines.flow.*

class AssetsSubscriptions(private val vertx: Vertx) {
    fun assetCreated(): Flow<AssetsMutation.AssetCreated> =
        vertx.eventBus()
            .localConsumer<AssetsMutation.AssetCreated>(AssetsMutation.MESSAGE_BUS_ASSET_CREATED)
            .bodyStream()
            .toReceiveChannel(vertx)
            .receiveAsFlow()
}

With the subscription being closed immediately, the application doesn't have any chance to actually write something to the bus.

Edit: after confirming separately that the Flow is set up and working correctly, here's some additional code which sets up the GraphQL instance. As you might have inferred from the snippet above, I'm using Kotlin and Vert.x.

   private val schemaGeneratorConfig: SchemaGeneratorConfig by lazy {
        SchemaGeneratorConfig(
            supportedPackages = listOf("com.marcusilgner.testing"),
            hooks = FlowSubscriptionSchemaGeneratorHooks(),
        )
    }

    private val graphQL: GraphQL by lazy {
        val queries = mutableListOf<TopLevelObject>()
        val mutations = mutableListOf<TopLevelObject>()
        val subscriptions = mutableListOf<TopLevelObject>()
        for (enhancement in pluginManager.getExtensions(GraphQLEnhancement::class.java)) {
            queries.addAll(enhancement.queries())
            mutations.addAll(enhancement.mutations())
            subscriptions.addAll(enhancement.subscriptions())
        }

        val schema = toSchema(
            schemaGeneratorConfig,
            queries = queries,
            subscriptions = subscriptions,
            mutations = mutations
        )
        GraphQL.newGraphQL(schema)
                .subscriptionExecutionStrategy(FlowSubscriptionExecutionStrategy())
                .build()
    }

    override suspend fun start() {
        val router = Router.router(vertx)
        router.route().handler(BodyHandler.create())

        val handlerOptions = GraphQLHandlerOptions()
            .setRequestMultipartEnabled(true)
            .setRequestBatchingEnabled(true)
        router.route("/graphql")
            .handler(GraphQLWSHandler.create(graphQL).connectionInitHandler {
                // no additional setup needed but if no init handler exists,
                // the handler won't send `ACK` to the client
                it.complete() 
            })
            .handler(GraphQLHandler.create(graphQL, handlerOptions))
        val iQlOptions = GraphiQLHandlerOptions().setEnabled(true)
        router.route("/graphiql/*").handler(GraphiQLHandler.create(iQlOptions))

        val httpServerOptions = HttpServerOptions()
            .addWebSocketSubProtocol("graphql-transport-ws")
        val httpServer = vertx.createHttpServer(httpServerOptions)
        val port = ConfigRetriever.create(vertx).config.await().getJsonObject("http").getInteger("port")
        httpServer.requestHandler(router).listen(port).await()
        logger.info("HTTP server now listening on $port")
    }

Everything else works already, only the flow-based subscriptions don't. I'll probably try writing some Publisher-based ones instead.

Update 2: I can confirm that when returning an instance of org.reactivestreams.Publisher instead, everything works as expected:

class AssetsSubscriptions(private val vertx: Vertx) {
    fun assetCreated(): Publisher<AssetsMutation.AssetCreated> {
        val readStream = vertx.eventBus()
            .localConsumer<AssetsMutation.AssetCreated>(AssetsMutation.MESSAGE_BUS_ASSET_CREATED)
            .bodyStream()
        val publisher = ReactiveWriteStream.writeStream<AssetsMutation.AssetCreated>(vertx)
        Pump.pump(readStream, publisher).also { it.start() }
        return publisher
    }
}