How to expose StateFlow but start and stop underlying Flow

user_5847275 0
Viewed 785

I have a class that receives data from the device sensors. I've wrapped the callback inside a callbackFlow in order to get a Flow.

private var sensorFlow: Flow<AccelerationState> = callbackFlow {
        val listener: SensorEventListener = object : SensorEventListener {
            override fun onSensorChanged(event: SensorEvent?) {
              ...
            }

            override fun onAccuracyChanged(sensor: Sensor?, accuracy: Int) {
              ...
            }
        }
        sensorManager.registerListener(listener, accelSensor, SensorManager.SENSOR_DELAY_NORMAL)

        awaitClose() {
            sensorManager.unregisterListener(listener, accelSensor)
        }
    }

Ideally want to expose a StateFlow so I've done this:

 sensorFlow.stateIn(
        scope = scope,
        started = WhileSubscribed(5000),
        initialValue = AccelerationState(0,0,0))

I want to be able to pause/resume the sensor readings which I would normally do by registering/unregistering the listener. My understanding is that I have to stop listening to the flow and then restart it again if I want to have that effect. I've tried using takeWhile { isGathering } but after the Flow is killed it can't be restarted again. And redeclaring the StateFlow does nothing as the rest of the app has the previous reference stored.

How can I expose a StateFlow but be able to change the underlying Flow from which it's created?

Edit: The StateFlow is then simply referenced in a ViewModel and can then be accessed by the View:

val sensorStateFlow = sensorDataRepository.sensorStateFlow
kotlin-flow kotlin-coroutines kotlin android
1 Answers

What if you were to use a flow of flows?

First, create a MutableStateFlow that will hold the "current" flow. Its initial value is just an empty flow.

val currentFlow = MutableStateFlow<Flow<T>>(emptyFlow())

Then, launch the StateFlow that subscribers will consume. This is based on a concatenated flow that always collects the latest flow that was emitted by currentFlow.

val stateFlow = currentFlow.flatMapLatest { it }.stateIn(scope)

Initially, the stateFlow is empty (and suspends, because it doesn't have an initial value). To start it, you can set the value of currentFlow:

currentFlow.value = createNewSensorFlow()

The values from the sensor flow will then be collected and available via the stateFlow. Setting a new value for currentFlow will cause the concatenated stateFlow to stop listening to (and terminate) the previous flow and start collecting from the new one instead. To pause the sensor you can set the value back to an empty flow:

currentFlow.value = emptyFlow()

You could still use SharingStarted.WhileSubscribed when launching the state flow, to ensure that the listeners aren't active when there are no subscribers. If you do that, the "current" flow will simply be collected again when new subscribers arrive. Because you're using a callback flow, that should work fine, and will just initialise the flow again to create a new listener.

As a simplification, if the sample rate is the only thing you need to change, consider something like this:

val sampleRate = MutableStateFlow<Int?>(null)
val stateFlow = sampleRate.flatMapLatest { sampleRate ->
    if (sampleRate == null) {
        emptyFlow()
    } else {
        createSensorFlow(sampleRate)
    }
}.stateIn(scope)

Here, instead of setting currentFlow, we just set sampleRate.value to whatever we want. Each time it's changed, the "current flow" (which is now created as part of flatMapLatest) is replaced. We set sampleRate.value to null to emit an empty flow and thereby pause the sensor.

user_46183310
Related