I have some code that takes items from a channel a...

# coroutines

[Dico]

I have some code that takes items from a channel and sends them in batches to a HTTP api with ktor.
How can I make it detect cancellation and send the remaining accumulated items in a clean way?
Pseudo code here:

private val processor = scope.launch {
        val batch = ArrayList<String>(MAX_BATCH)
        
        suspend fun sendBatch() {
            val payload = batch.joinToString(separator = "\n")
            batch.clear()
            return sendPayload(payload)
        }
        
        try {
            while (isActive) {
                batch.add(hitChannel.receive())
                if (batch.size == MAX_BATCH) {
                    sendBatch()
                }
            }
        } finally {
            if (batch.isNotEmpty()) {
                sendBatch()
            }
        }
    }

I expect that final sendBatch will throw exception because coroutine no longer isActive. I can't do withContext(NonCancellable) either at that point.

[Dominaezzz]

You want to send remaining items inside the batch? or batches after the batch?

[Dico]

remaining items inside the batch

[Dominaezzz]

Ah, if hitChannel.receive() fails, you want to call sendBatch?

[Dominaezzz]

What's being cancelled here?

[Dico]

The scope is being cancelled, sorry for lack of clarity

[Dico]

that means receive() will fail I think

[Dico]

Perhaps I should test it with a modified sendBatch

[Dominaezzz]

I would suggest a refactor.

[Dico]

you mean start from scratch with the logic?

[Dominaezzz]

The processor and the batching should have different lifecycles.

[Dico]

I see what you mean!

[Dico]

so how would you implement the batching on its own lifecycle?

[Dico]

I suppose I should close the channel and then detect that, that makes sense

[Dominaezzz]

Something like this https://kotlinlang.slack.com/archives/C1CFAFJSK/p1563798636010800

[Dominaezzz]

Abstract away the batching from the request sender caller.

[Dominaezzz]

Basically pass the request to another coroutine which will be cancelled differently from the processor.

[Dominaezzz]

Closing the channel also works. 😅

[Dico]

Thanks for the help

[Dico]

For reference - the code becomes

private val processor = scope.launch {
        val batch = ArrayList<String>(MAX_BATCH)

        suspend fun sendBatch() {
            val payload = batch.joinToString(separator = "\n")
            batch.clear()
            return sendPayload(payload)
        }

        for (item in hitChannel) {
            batch.add(hitChannel.receive())
            if (batch.size == MAX_BATCH) {
                sendBatch()
            }
        }

        if (batch.isNotEmpty()) {
            sendBatch()
        }
        
        scope.cancel()
    }

    fun stop() {
        hitChannel.close()
    }

[Dominaezzz]

You removed the try/finally?

[Dico]

I don't expect errors

[Dico]

If there are errors they will be reported by the uncaught exception handler

[Dico]

And the data handled here is just for google analytics so I'm not fussed

[Dico]

In hindsight, it should probably be there

[Dominaezzz]

If hitChannel is closed before MAX_BATCH is attained, then some items won't be sent, even though they've gone through the channel.

[Dico]

This is the only coroutine receiving from the channel

[Dico]

I'm using a for loop, so receive() will never fail

[Dominaezzz]

Oh. You want batch.add(item)

[Dico]

What do you mean?

[Dominaezzz]

Instead of hitChannel.receive().

[Dominaezzz]

for loop does receive for you.

[Dico]

right, I'm an idiot.

[Dico]

thx

[Dominaezzz]

No problem. This was fun!