Sometimes, ConflatedBroadcastChannel fires recent value without any action

Question

In Google's official codelab about advanced-coroutines-codelab sample, they've used ConflatedBroadcastChannel to watch a variable/object change.

I've used the same technique in one of my side projects, and when resuming the listening activity, sometimes ConflatedBroadcastChannel fires it's recent value, causing the execution of flatMapLatest body without any change.

I think this is happening while the system collects the garbage since I can reproduce this issue by calling System.gc() from another activity.

Here's the code

MainActivity.kt

class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        val viewModel = ViewModelProvider(this).get(MainViewModel::class.java)
        val tvCount = findViewById<TextView>(R.id.tv_count)

        viewModel.count.observe(this, Observer {
            tvCount.text = it
            Toast.makeText(this, "Incremented", Toast.LENGTH_LONG).show();
        })

        findViewById<Button>(R.id.b_inc).setOnClickListener {
            viewModel.increment()
        }

        findViewById<Button>(R.id.b_detail).setOnClickListener {
            startActivity(Intent(this, DetailActivity::class.java))
        }

    }
}

MainViewModel.kt

class MainViewModel : ViewModel() {

    companion object {
        val TAG = MainViewModel::class.java.simpleName
    }

    class IncrementRequest

    private var tempCount = 0
    private val requestChannel = ConflatedBroadcastChannel<IncrementRequest>()

    val count = requestChannel
        .asFlow()
        .flatMapLatest {
            tempCount++
            Log.d(TAG, "Incrementing number to $tempCount")
            flowOf("Number is $tempCount")
        }
        .asLiveData()

    fun increment() {
        requestChannel.offer(IncrementRequest())
    }
}

DetailActivity.kt

class DetailActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_detail)
        val button = findViewById<Button>(R.id.b_gc)


        val timer = object : CountDownTimer(5000, 1000) {
            override fun onFinish() {
                button.isEnabled = true
                button.text = "CALL SYSTEM.GC() AND CLOSE ACTIVITY"
            }

            override fun onTick(millisUntilFinished: Long) {
                button.text = "${TimeUnit.MILLISECONDS.toSeconds(millisUntilFinished)} second(s)"
            }
        }

        button.setOnClickListener {
            System.gc()
            finish()
        }

        timer.start()

    }
}

Here's the full source code : CoroutinesFlowTest.zip

• Why is this happening?
• What am I missing?

Answer 1

Quoting from the official response, (The simple and straightforward solution)

The problem here is that you are trying to use ConflatedBroadcastChannel for events, while it is designed to represent current state as shown in the codelab. Every time the downstream LiveData is reactivated it receives the most recent state and performs the incrementing action. Don't use ConflatedBroadcastChannel for events.

To fix it, you can replace ConflatedBroadcastChannel with BroadcastChannel<IncrementRequest>(1) (non-conflated channel, which is Ok for events to use) and it'll work as you expect it too.

Answer 2

In addition to the answer of Kiskae:

This might not be your case, but you can try to use BroadcastChannel(1).asFlow().conflate on a receiver side, but in my case it led to a bug where the code on a receiver side didn't get triggered sometimes (I think because conflate works in a separate coroutine or something).

Or you can use a custom version of stateless ConflatedBroadcastChannel (found here).

class StatelessBroadcastChannel<T> constructor(
    private val broadcast: BroadcastChannel<T> = ConflatedBroadcastChannel()
) : BroadcastChannel<T> by broadcast {

    override fun openSubscription(): ReceiveChannel<T> = broadcast
        .openSubscription()
        .apply { poll() }

}

Answer 3

On Coroutine 1.4.2 and Kotlin 1.4.31

Without using live data

private var tempCount = 0
private val requestChannel = BroadcastChannel<IncrementRequest>(Channel.CONFLATED)

val count = requestChannel
        .asFlow()
        .flatMapLatest {
            tempCount++
            Log.d(TAG, "Incrementing number to $tempCount")
            flowOf("Number is $tempCount")
        }

Use Flow and Coroutine

lifecycleScope.launchWhenStarted {
     viewModel.count.collect {
          tvCount.text = it
          Toast.makeText(this@MainActivity, "Incremented", Toast.LENGTH_SHORT).show()
    }
}

Without using BroadcastChannel

private var tempCount = 0
    private val requestChannel = MutableStateFlow("")

    val count: StateFlow<String> = requestChannel
    
    fun increment() {
        tempCount += 1
        requestChannel.value = "Number is $tempCount"
    }

Answer 4

The reason is very simple, ViewModels can persist outside of the lifecycle of Activities. By moving to another activity and garbagecollecting you're disposing of the original MainActivity but keeping the original MainViewModel.

Then when you return from DetailActivity it recreates MainActivity but reuses the viewmodel, which still has the broadcastchannel with a last known value, triggering the callback when count.observe is called.

If you add logging to observe the onCreate and onDestroy methods of the activity you should see the lifecycle getting advanced, while the viewmodel should only be created once.