roscrazy

override fun singleLocation(): Single<Location> = observeLocationChange()
        .firstOrError()
        .timeout(locationAttributes.requestTimeOut, TimeUnit.MILLISECONDS)

roscrazy

fusedLocationService.requestLocationUpdates(locationAttributes)
            .onErrorResumeNext { it: Throwable ->
                androidLocationService.requestLocationUpdates(locationAttributes)
            }

roscrazy

internal interface LocationService {
    fun requestLocationUpdates(attributes: RxLocationAttributes): Observable<Location>
}

roscrazy

internal class RxLocationManagerImpl(
    private val fusedLocationService: LocationService,
    private val androidLocationService: LocationService,
    private val locationAttributes: RxLocationAttributes
) : RxLocationManager {
    private val sharedLocationObservable = createLocationObservable()

    override fun singleLocation(): Single<Location> = observeLocationChange()
        .firstOrError()
        .timeout(locationAttributes.requestTimeOut, TimeUnit.MILLISECONDS)

roscrazy

internal class AndroidLocationService(private val locationManager: LocationManager) :
    LocationService {
    override fun requestLocationUpdates(attributes: RxLocationAttributes): Observable<Location> {
        return Observable.create<Location> { emitter ->
            log { "AndroidLocationService ObservableOnSubscribe create" }
            val locationListener = RxLocationListener(emitter)
            emitter.setCancellable {
                log { "AndroidLocationService ObservableOnSubscribe canceled" }
                locationManager.removeUpdates(locationListener)
            }

roscrazy

override fun requestLocationUpdates(attributes: RxLocationAttributes): Observable<Location> {
    return createLocationObservable(attributes)
        .retry { n: Int, e: Throwable ->
            (n < attributes.retryAttempt
                    && e !is GooglePlayServicesNotAvailableException
                    && e !is SecurityException)
        }
}

roscrazy

emitter.setCancellable(getCancellable(listener))

...
private fun getCancellable(locationListener: LocationCallback): Cancellable {
    return Cancellable {
        log { "FusedLocationService Observable has canceled" }
        fusedLocationProviderClient.removeLocationUpdates(locationListener)
    }
}

roscrazy

private fun createLocationObservable(attributes: RxLocationAttributes): Observable<Location> {
    return Observable.create { emitter ->
        val listener = getLocationListener(emitter)
        val completeListener = getOnCompleteListener(emitter)
        try {
            fusedLocationProviderClient.lastLocation.addOnSuccessListener {
                if (!emitter.isDisposed && it != null) emitter.onNext(it)
            }

roscrazy

internal class FusedLocationService(
    private val fusedLocationProviderClient: FusedLocationProviderClient
) : LocationService {
    override fun requestLocationUpdates(attributes: RxLocationAttributes): Observable<Location> {
        return getLocationObservable(attributes)
            .retry { n: Int, e: Throwable ->
                (n < attributes.retryAttempt
                        && e !is GooglePlayServicesNotAvailableException
                        && e !is SecurityException)
            }