datasource-SharedFlow.kt

@Singleton
class NearbyUsersDataSource @Inject constructor() {
    // Ideally, those should be constructor-injected.
    val geoFire = GeoFire(FirebaseDatabase.getInstance().getReference("geofire"))
    val geoLocation = GeoLocation(0.0, 0.0)
    val radius = 100.0
    
    val geoQuery = geoFire.queryAtLocation(geoLocation, radius)
    
    private fun GeoQuery.asFlow() = callbackFlow {
        val listener: GeoQueryEventListener = object : GeoQueryEventListener {
            val map = mutableMapOf<Key, GeoLocation>()
            override fun onKeyEntered(key: String, location: GeoLocation) {
                map[key] = location
            }
            override fun onKeyExited(key: String) {
                map.remove(key)
            }
            override fun onKeyMoved(key: String, location: GeoLocation) {
                map[key] = location
            }
            override fun onGeoQueryReady() {
                emit(State.Ready(map.toMap())
            }
            override fun onGeoQueryError(e: DatabaseError) {
                emit(State.Error(map.toMap(), e.toException())
            }
        }
        
        addGeoQueryEventListener(listener)
        
        awaitClose { removeGeoQueryEventListener(listener) }
    }.shareIn(
         ProcessLifecycleOwner.get().lifecycleScope,
         SharingStarted.WhileSubscribed(),
         1
    )

    val locations: Flow<State> = geoQuery.asFlow()
                     
    sealed class State(open val value: Map<Key, GeoLocation>) {
        data class Ready(
            override val value: Map<Key, GeoLocation>
        ) : State(value)
        
        data class Error(
            override val value: Map<Key, GeoLocation>,
            val exception: Exception
        ) : State(value)
    }
}