Kotlin Flows, executing code when there are no active subscribers

Issue

I have created a location repository that is responsible for collecting location data and transmitting it to whoever listens, the code right now looks something like this:

@Singleton
class LocationRepository @Inject constructor(
    app: Application
) {

    private val _location = MutableStateFlow<Location?>(null)
    private val subscribers = AtomicInteger()
    val location: Flow<Location?> = _location.onSubscription {
        if (subscribers.getAndIncrement() == 0)
            startUpdates()
    }.onCompletion {
        if (subscribers.decrementAndGet() == 0)
            stopUpdates()
    }

    private val client = LocationServices.getFusedLocationProviderClient(app)
    private val locCb = object : LocationCallback() {
        override fun onLocationResult(var1: LocationResult?) {
            newLocationResult(var1)
        }
        override fun onLocationAvailability(var1: LocationAvailability) {}
    }
    private val request = LocationRequest.create()

    init {
        request.priority = LocationRequest.PRIORITY_BALANCED_POWER_ACCURACY
        request.interval = 5000L
        request.fastestInterval = 2500L
    }


    private fun startUpdates() { //code to start getting location updates }
    private fun stopUpdates() { //code to stop getting location updates }

}

right now I am handling the current number of subscribers with an AtomicInteger, is there a way to run the startUpdates code and the stopUpdates code automatically when the first subscriber subscribes and the last subscriber unsubscribes?

Solution

Please check subscriptionCount property of MutableSharedFlow, it may be what you are looking for. It has type of StateFlow<Int> and emits the number of subscribers (active collectors) to this shared flow. Example:

_location.subscriptionCount
    .map { count -> count > 0 } // map count into active/inactive flag
    .distinctUntilChanged() // only react to true<->false changes
    .onEach { isActive -> // configure an action
        if (isActive) startUpdates() else stopUpdates()
    }
    .launchIn(scope) // launch it

Answered By - Sergey