Create a Flow of location updates on Android (using kotlinx.coroutines), backed by Fused Location Provider from Google Play Services.

FusedLocationFlow.kt

/*
 * Copyright 2019 Louis Cognault Ayeva Derman. Use of this source code is governed by the Apache 2.0 license.
 */

import android.location.Location
import com.google.android.gms.location.LocationCallback
import com.google.android.gms.location.LocationRequest
import com.google.android.gms.location.LocationResult
import com.google.android.gms.location.LocationServices
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.tasks.await
import splitties.init.appCtx // Similar to Firebase auto-initialization of applicationContext,
// usages can be replaced with Context receiver or parameter if you really enjoy parameter passing.

@ExperimentalCoroutinesApi
@Throws(SecurityException::class)
inline fun fusedLocationFlow(
    configLocationRequest: LocationRequest.() -> Unit
): Flow<Location> = fusedLocationFlow(
    locationRequest = LocationRequest.create().apply(configLocationRequest)
)

@Throws(SecurityException::class)
@ExperimentalCoroutinesApi
fun fusedLocationFlow(
    locationRequest: LocationRequest
): Flow<Location> = channelFlow {
    val locationClient = LocationServices.getFusedLocationProviderClient(appCtx)
    val locationCallback = object : LocationCallback() {
        override fun onLocationResult(result: LocationResult) {
            result.locations.forEachByIndex { offerCatching(it) }
        }
    }
    locationClient.lastLocation.await<Location?>()?.let { send(it) }
    locationClient.requestLocationUpdates(locationRequest, locationCallback, null).await()
    awaitClose {
        locationClient.removeLocationUpdates(locationCallback)
    }
}

/**
 * [SendChannel.offer] that returns `false` when this [SendChannel.isClosedForSend], instead of
 * throwing.
 *
 * [SendChannel.offer] throws when the channel is closed. In race conditions, especially when using
 * multithreaded dispatchers, that can lead to uncaught exceptions as offer is often called from
 * non suspending functions that don't catch the default [CancellationException] or any other
 * exception that might be the cause of the closing of the channel.
 */
// Copy pasted from splitties.coroutines
fun <E> SendChannel<E>.offerCatching(element: E): Boolean {
    return runCatching { offer(element) }.getOrDefault(false)
}


/**
 * Iterates the receiver [List] using an index instead of an [Iterator] like [forEach] would do.
 * Using this function saves an [Iterator] allocation, which is good for immutable lists or usages
 * confined to a single thread like UI thread only use.
 * However, this method will not detect concurrent modification, except if the size of the list
 * changes on an iteration as a result, which may lead to unpredictable behavior.
 *
 * @param action the action to invoke on each list element.
 */
// Copy pasted from splitties.collections
inline fun <T> List<T>.forEachByIndex(action: (T) -> Unit) {
    val initialSize = size
    for (i in 0..lastIndex) {
        if (size != initialSize) throw ConcurrentModificationException()
        action(get(i))
    }
}

Hm, I have done something very similar and it seems I can't get the awaitClose to fire when launched with lifecycleScope. Did you verify it?

EDIT: Sorry my bad, I didn't use viewLifecycleOwner.lifecycleScope. A quick addition as well. Multiple calls to this util would create multiple location requests correct? It would seem a bit heavy. Having it exposed as a property, wrapping this in a singleton-injectable-class and sharing the property would be more ideal no? I know that the Flow.share() is not released yet but what are your thoughts ? =)

@sp00ne I'm using toLiveData from AndroidX Lifecycle LiveData KTX (which you can convert back to flow after keeping the same reference for the middle sharing livedata). The timeout is 5 seconds by default.

@LouisCAD Won't making it a LiveData inbetween cause exceptions not to be handled? If you look at the source for .asFlow() it doesn't pass the exception:

fun <T> LiveData<T>.asFlow(): Flow<T> = flow {
    val channel = Channel<T>(Channel.CONFLATED)
    val observer = Observer<T> {
        channel.offer(it)
    }
    withContext(Dispatchers.Main.immediate) {
        observeForever(observer)
    }
    try {
        for (value in channel) {
            emit(value)
        }
    } finally {
        GlobalScope.launch(Dispatchers.Main.immediate) {
            removeObserver(observer)
        }
    }
}

Or am I missing something? I.e if you do this:

liveDataThing.asFlow()
         .catch {
               Log.d("Error caught")
         }
         .launchIn(someScope)

and LiveData receives an error -> "catch" won't be reached, rather the scope will receive the error.

@sp00ne You need to handle the errors before converting to a LiveData.

@LouisCAD i.e "catch" is not going to fire downstream making it sort of invalid for attempts of retrying or handling it in a "Flow"-manner 🤷

I wish you could update it. Appreciate so much. Thank you.

locationClient.requestLocationUpdates(locationRequest, locationCallback, null).await()

The await call seems redundant, since the awaitClose block will keep the channel open.

i think so, i already made this class today and it worked without await()