Reactive extensions for GoogleApiClient using Kotlin/RxJava

ExampleUse.kt

/* Example use of the GoogleApiClient reatcive extions sketched out below and how we can use them to
* 1. Create an api client
* 2. Make a couple of serial api calls
* 3. Clean up the api client
*/

  // Let's create a refernce to a lambda we can use to clean up the client when we are finished with it.
  var destroy: (() -> Unit)? = null

  //Create our api client with the APIs we need to use
  GoogleApiClientUtil.get(
      context,
      listOf(
          Pair(Auth.CREDENTIALS_API, null)
      ),
      { destroy = it; false }
  ).map { client ->
    //First Api call
    Auth.CredentialsApi.save(client, credential).observable
        .flatMap { result ->
          //Second Api call
          Auth.CredentialsApi.someOtherApi(client, result.something).observable
        }
        //Clean up the client on compltion of all our api calls
        .doOnCompleted { destroy?.invoke() }
        .subscribe { result ->
          // Do something with the result
        }
  }

GoogleApiRxUtils.kt

import android.os.Bundle
import android.support.v4.app.FragmentActivity
import com.google.android.gms.common.api.Api
import com.google.android.gms.common.api.GoogleApiClient
import com.google.android.gms.common.api.PendingResult
import com.google.android.gms.common.api.Result
import rx.Emitter
import rx.Observable
import java.util.concurrent.atomic.AtomicInteger

val <R : Result> PendingResult<R>.observable: Observable<R>
  get() = Observable.fromEmitter<R>({ emitter ->
    setResultCallback { result ->
      emitter.onNext(result)
      emitter.onCompleted()
    }
  }, Emitter.BackpressureMode.BUFFER)

class GoogleApiClientUtil {
  companion object {
    val clientCount by lazy { AtomicInteger() }

    fun get(context: FragmentActivity, apis: List<Pair<Api<*>, Api.ApiOptions.HasOptions?>>, shouldDestroy:((() -> Unit) -> Boolean)? = null): Observable<GoogleApiClient> {

      val clientId = clientCount.incrementAndGet()
      return Observable.fromEmitter<GoogleApiClient>({ emitter ->
        val builder = GoogleApiClient.Builder(context)
            .enableAutoManage(context, clientId, { emitter.onError(RuntimeException("Connection failed")) })

        apis.forEach @Suppress("UNCHECKED_CAST") a@ {
          if (it.second == null) {
            builder.addApi(it.first as Api<Api.ApiOptions.NotRequiredOptions>)
          } else {
            builder.addApi(it.first as Api<Api.ApiOptions.HasOptions>, it.second!!)
          }
          return@a
        }

        val client = builder.build()

        val connectionCallbacks = object : GoogleApiClient.ConnectionCallbacks {
          override fun onConnected(p0: Bundle?) {
            emitter.onNext(client)
            emitter.onCompleted()
          }

          override fun onConnectionSuspended(p0: Int) {
            emitter.onError(RuntimeException("Connection failed"))
          }
        }

        client.registerConnectionCallbacks(connectionCallbacks)

        val destroy:() -> Unit = {
          client.unregisterConnectionCallbacks(connectionCallbacks)
          client.stopAutoManage(context)
        }

        emitter.setCancellation {
          if(shouldDestroy?.invoke(destroy) != false){
            destroy()
          }
        }
      }, Emitter.BackpressureMode.BUFFER)

    }
  }
}