RxJava2 how to separate different implementation of observable emitter

This is my situation. I want to expose 2 different implementation of Observable<'Location'> from my native LocationManager of Android or from the Google Services.

I want to check if I use the native approach or gms.

So in the end I want to expose Observable<Location> to my client - he doesn't need to know from which approach I gathered the location. NOTE that I am using this library: https://github.com/mcharmas/Android-ReactiveLocation

for exposing Observable from google services. It already expose the Observable that I am searching for. But what about the other - Location Manager. It uses callbacks.

Here is my implementation:

  var locationEmitter : Observable<Location> = Observable.empty()


init {
    configureEmitter()
}

@SuppressLint("MissingPermission")
private fun configureEmitter(){
    if (!isUsingLocationNativeApi)
      locationEmitter = reactiveLocationProvider.getUpdatedLocation(reactiveLocationRequest)
    else{
      configureNativeLocationEmitter()
    }
}

@SuppressLint("MissingPermission")
private fun configureNativeLocationEmitter() {


    val mLocationCallbackNativeApi: LocationListener = object : LocationListener {

        override fun onLocationChanged(location: Location) {
        locationEmitter = Observable.create<Location> { emitter -> emitter.onNext(location) }

        }

        override fun onStatusChanged(provider: String, status: Int, extras: Bundle) {}

        override fun onProviderEnabled(provider: String) {}

        override fun onProviderDisabled(provider: String) {}

    }


    try {
        locationManager.requestLocationUpdates(LocationManager.GPS_PROVIDER,
                (geoEventsIntervalSeconds * 1000).toLong(),
                geoEventsDistanceMeters.toFloat(),
                mLocationCallbackNativeApi,
                Looper.getMainLooper())
    } catch (ignored: IllegalArgumentException) {
        ignored.printStackTrace()

    }

    try {
        locationManager.requestLocationUpdates(LocationManager.NETWORK_PROVIDER,
                (geoEventsIntervalSeconds * 1000).toLong(),
                geoEventsDistanceMeters.toFloat(),
                mLocationCallbackNativeApi,
                Looper.getMainLooper())
    } catch (ignored: IllegalArgumentException) {
        ignored.printStackTrace()
    }
}



@SuppressLint("MissingPermission")
override fun onLocationUpdate(): Observable<Location> {
    return locationEmitter
    }

}

However, it does not work for the LocationManager implementation,

In the client side I want to be able to do something like this:

 rxLocationRepository.onLocationUpdate()
            .subscribe(Consumer { location -> 
    //do something with location, no matter it is from LocationManager or google play services
 })

How to do that in Kotlin?

UPDATE

How to get rid of the tracking from subject(actually get rid of subject) and the other locationProvider? I don't want to lose use them if the client will not want to use this and for example invoke such method in the client side

 override fun stopLocationUpdates() {
 //get rid of the location updates
}

E.g. in the client side:

 rxLocationRepository.stopLocationUpdates()

Solution

You are creating a new observable every time onLocationChanged is called, which makes no sense. Just use a PublishSubject (or BehaviorSubject may be more suitable in this case) which can be returned as Observable to clients and call subject.onNext(location) in onLocationChanged.