Streaming CSV back to API client

[dodalovicgran]

Hi, i have a case where I need to serve CSVs to my clients, which can be hundreds of megabytes in size.
I'm using Kotlin, coroutines, and I was wondering: can I use Kotlin Flows somehow for that?

Some pseudo code snippet would be nice.

I was thinking of something like this, but kotlin friendly:

    @Path("/csv")
    @GET
    @Produces("text/csv")
    fun getCSV(): Multi<String> {
        return db.query("SELECT * FROM time_entry")
            .execute()
            .toMulti()
            .onItem()
            .transform { rowSet ->
                rowSet.joinToString("\n") { row ->
                    """${row.getUUID("id")},${row.getUUID("client_id")},${row.getString("description")},${
                        row.getOffsetDateTime(
                            "started_on"
                        )
                    },${row.getOffsetDateTime("completed_on")}"""
                }
            }
    }

[quarkus-bot[bot]]

/cc @evanchooly (kotlin), @geoand (kotlin)

[geoand]

I personally don't have any experience with Kotlin Flow :(

[mschorsch]

#18222 😁

[mschorsch]

Yes, you can 😃, Resteasy Reactive has built in support for Kotlin Flow.

You have the following options:

1. Replace the return type Multi with Flow and use Flow as described in the Kotlin docs (https://kotlinlang.org/docs/flow.html).
2. Replace the return type Multi with Flow and convert Multi to Flow with asFlow (https://smallrye.io/smallrye-mutiny/latest/guides/kotlin/#processing-a-multi-as-flow). For this you need to add an additional dependency mutiny-kotlin.

[mschorsch]

Unfortunately, the Kotlin Flow support is not documented.

[geoand]

Would you like to contribute this documentation?

[mschorsch]

I will give it a try

[dodalovicgran]

Yes, you can 😃, Resteasy Reactive has built in support for Kotlin Flow.

You have the following options:

1. Replace the return type Multi with Flow and use Flow as described in the Kotlin docs (https://kotlinlang.org/docs/flow.html).
2. Replace the return type Multi with Flow and convert Multi to Flow with asFlow (https://smallrye.io/smallrye-mutiny/latest/guides/kotlin/#processing-a-multi-as-flow). For this you need to add an additional dependency mutiny-kotlin.

@mschorsch Thank you for the prompt response! I really appreciate it. One quick question: does this mean that the single item will be pumped into response as soon as I prepare it (transform function from above completes)?

[mschorsch]

Yes, this should be the case, but beware, if you write a lot of data very quickly then you should define a suitable backpressure strategy otherwise you will either run out of memory or some of the CSV data will be dropped (see bufferCapacity and bufferOverflowStrategy).

There's an important note in the Mutiny-Guide (https://smallrye.io/smallrye-mutiny/latest/guides/kotlin/#processing-a-multi-as-flow)

There’s no flow control availabe for Kotlin’s Flow. Published items are buffered for consumption using a coroutine Channel. The buffer size and overflow strategy of that Channel can be configured using optional arguments: Multi.asFlow(bufferCapacity = Channel.UNLIMITED, bufferOverflowStrategy = BufferOverflow.SUSPEND), for more details please consult the method documentation.

One more note: Query db.query("SELECT * FROM time_entry") loads all data into memory.

[dodalovicgran]

I guess, the only way is to paginate on data and use the buffer you mentioned ....

[mschorsch]

I have not tested it, but something like that should do the trick:

import io.smallrye.mutiny.coroutines.awaitSuspending
import io.vertx.mutiny.pgclient.PgPool
import io.vertx.mutiny.sqlclient.*
import jakarta.enterprise.context.ApplicationScoped
import jakarta.ws.rs.GET
import jakarta.ws.rs.Path
import jakarta.ws.rs.Produces
import jakarta.ws.rs.core.MediaType
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.flow

@Path("/fruits")
@ApplicationScoped
class FruitsResource(
    private val pgPool: PgPool
) {

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    fun getFruits(): Flow<String> {
        return flow { getFruits() }
    }

    private suspend fun FlowCollector<String>.getFruits() {
        pgPool.withTransactionAwait {
            val pq = prepare("SELECT name from fruits").awaitSuspending()
            val cursor = pq.cursor()
            while (cursor.hasMore()) {
                cursor.read(50).awaitSuspending().forEach {
                    emit(it.getString(1))
                }
            }
        }
    }

    private suspend fun <T> PgPool.withTransactionAwait(block: suspend SqlConnection.() -> T) {
        val sqlConnection = connection.awaitSuspending()
        try {
            val transaction = sqlConnection.begin().awaitSuspending()
            try {
                block(sqlConnection)
                transaction.commit().awaitSuspending()
            } catch (ex: Exception) {
                transaction.rollback().awaitSuspending()
            }
        } finally {
            sqlConnection.close().awaitSuspending()
        }
    }
}