I'm trying to return a large json array using kotl...

# ktor

[dave08]

I'm trying to return a large json array using kotlinx serialization, but I'm getting Java heap errors... is there a way to just return a Flow (coroutines) and have Ktor handle gathering the result into a streamed json response?

[Aleksei Tirman [JB]]

Unfortunately, Ktor doesn't support Flow as the type of a response object. How big is the JSON array?

[dave08]

More than 3 thousand objects...

[Aleksei Tirman [JB]]

You can try to respond with an OutputStream to avoid loading the entire JSON into memory:

call.respondOutputStream(contentType = ContentType.Application.Json) {
    Json.encodeToStream(123, this)
}

[dave08]

Is there a way to access my content resolver's Json object from a route or do I need to create a new one?

[Aleksei Tirman [JB]]

Do you mean a ContentConverter ?

[dave08]

I meant ContentNegotiation's Kotlinx serializer instance...

[dave08]

I could try to save it elsewhere and pass it with DI, but it would have been convenient to have it available in Ktor

[dave08]

I usually don't need it, but this is a funny case...

[Aleksei Tirman [JB]]

Unfortunately, you cannot extract one passed to the ContentNegotiation plugin.

[dave08]

@Aleksei Tirman [JB] For some funny reason, I tried:

call.respondOutputStream(contentType = ContentType.Application.Json) {
                    repository.getFoo(lastSync).collect { foo -> json.encodeToStream(foo, this) }
                }

But I'm getting reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response from Intellij's http scratch file when trying out that endpoint...

[dave08]

Maybe I shouldn't be doing the collection inside that block, but it seems like it's a suspend fun, so why not? Unless I need to call something when I'm finished collecting? Or maybe I'm supposed to add [ and ] and , to build the complete json?

[dave08]

Ok, it seems like this did the trick:

suspend inline fun <reified T : Any> RoutingCall.respondFlowToJsonArray(flow: Flow<T>) {
    respondOutputStream(contentType = ContentType.Application.Json) {
        sink().buffer().use { sink ->
            sink.writeUtf8("[")
            var isFirst = true
            flow.collect { obj ->
                if (!isFirst)
                    sink.writeUtf8(",")
                else isFirst = false

                Json.encodeToBufferedSink(obj, sink)
            }
            sink.writeUtf8("]")
        }
    }
}

It might be nice to add this to Ktor though...

[dave08]

Nope, it worked in my unit test with a bit of data, but not on production where i got the same error...

[Aleksei Tirman [JB]]

Does the flow collection work outside of the respondOutputStream 's block?

[dave08]

How would I do that w/o first saving all those classes into a List, which would then be much less efficient for what I'm trying to acheive @Aleksei Tirman [JB]? Unless I didn't understand your question?

[Aleksei Tirman [JB]]

Actually, you can write JSON objects directly to the ByteChannel and stream the response by returning the channel. Here is an example:

embeddedServer(Netty, port = 8080) {
    val f = flow {
        emit("Hello world")
        delay(1000)
        emit("Hello world")
        delay(1000)
        emit("Hello world")
        delay(1000)
    }

    routing {
        get {
            val channel = ByteChannel()

            launch {
                f.collect { foo ->
                    channel.writeStringUtf8(Json.encodeToString(foo))
                }

                channel.close()
            }

            call.respond(channel)
        }
    }

}.start(wait = true)

[dave08]

Is that much better than the code I posted? Could that maybe solve the problem (which I have no clue what it could be, since I don't get ANY logs on the server side, not a crash nor even a log of the request itself...)?

[Aleksei Tirman [JB]]

The above code should solve the problem with the PrematureCloseException and in theory, it should be more performant because reading from the OutputStream blocks the thread.

[dave08]

Well, I put in some logs, it got to 8310 items processed, but didn't print the last log... it just prints the first log 3 times and it seems to give that error... here's what I did:

suspend inline fun <reified T : Any> RoutingCall.respondFlowToJsonArray2(flow: Flow<T>) {
    val channel = ByteChannel()
    launch {
        val logger = application.log

        <http://logger.info|logger.info>("Starting to respond flow to json array")
        channel.writeStringUtf8("[")
        var isFirst = true
        flow.collectIndexed { index, obj ->
            if (index % 15 == 0) <http://logger.info|logger.info>("Responding flow item $index")
            if (!isFirst)
                channel.writeStringUtf8(",")
            else isFirst = false
            channel.writeStringUtf8(Json.encodeToString(obj))
        }
        channel.writeStringUtf8("]")
        <http://logger.info|logger.info>("Finished responding flow to json array")

        channel.close()
    }

    respond(channel)
}

[dave08]

When I look at the metrics in prometheus, I see that it doesn't reach it's limit (2gb) it get to about 1.2gb while processing this...