Properly finalize flow from callbackFlow

Question

In the example below, I simply wrote a callbackflow example.

    var listener: ((String) -> Unit)? = null
    fun callback1() {
        listener?.let{
            it("1")
        }
    }

    fun callback2() {
        listener?.let{
            it("2")
        }
    }

    fun fromRepository(): Flow<String> = callbackFlow {
        listener = { result->
            trySend(result)
        }
        awaitClose {
            listener = null
        }
    }

    fun test() {
        val job = lifecycleScope.launch {
            fromRepository()
                .onCompletion { Timber.d("Completed") }
                .collectLatest {
                    Timber.d("Number: $it")
                }
        }
        callback1()
        callback2()
        callback1()
    }

I can see 1,2,1 output on the log as a result of the example code above. but even if I set listener = null the job continues to run. I edited the test function a bit for this.

    fun test() {
        val job = lifecycleScope.launch {
            fromRepository()
                .onCompletion { Timber.d("Completed") }
                .collectLatest {
                    Timber.d("Number: $it")
                }
        }
        callback1()
        callback2()
        callback1()
        listener = null

        lifecycleScope.launch(Dispatchers.IO) {
            delay(1000)
            Timber.d("job status: ${job.isActive}")
        }
    }

In addition to the above output, I can now see job status: true as well.

How can I completely finish the flow in the example above. How can I be protected from memory leak in this situation?

                .collectLatest {
                    Timber.d("Number: $it")
                    cancel()
                }

If I use the cancel function, it happens as I want, both the job becomes false and the onCompletion function calls. Is there any other solution without using the Cancel function?

Answer 1

How can I completely finish the flow in the example above. How can I be protected from memory leak in this situation?

I guess it depends on what you're trying to do. A flow built with callbackFlow can be completed in several ways.

On the producer side, you can close() the backing channel (from inside callbackFlow), and this will have the effect of completing the flow from the consumer's point of view (a collect call will complete and return).

On the consumer side, you can cancel the collection of the flow. This can be done by using operators such as first() or take(5) that will stop the collection without waiting for the flow to actually complete. It can also be done by cancelling the coroutine in which the collect call happens.