Obteniendo una IllegalStateException al usar la API People de Google a pesar de suscribirse a otro hilo usando RxJava

Nota: Estoy usando Kotlin junto con RxKotlin para las útiles funciones de extensión.

Intento get algunos datos de usuario usando la API People de Google, y acabo de migrar todas mis AsyncTasks y cosas para usar RxJava. Todo funciona, excepto por esta parte:

private fun getGooglePerson(service: PeopleService?, account: GoogleSignInAccount) = Single.just(service?.people() ?.get("people/me") ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers") ?.execute() ?: Person()) // If null, return an empty person (shouldn't ever be null, though) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeBy( onError = {/*Do something*/}, onSuccess = {/*Do something*/} ) 

Específicamente, recibo una IllegalStateException con el post, Calling this from your main thread can lead to deadlock . Estoy confundido porque tenía la printing de que llamar a subscribeOn debería desplazar todo, tanto por encima como por debajo (a less que lo observeOn ) a otro hilo (en este caso, Schedulers.io() ) intenté reescribir el código de esta manera:

 private fun getGooglePerson(service: PeopleService?, account: GoogleSignInAccount) = service?.people() ?.get("people/me") ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers") ?.execute() ?: Person() // If null, return an empty person (shouldn't ever be null, though) .toSingle() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeBy( onError = {/*Do something*/}, onSuccess = {/*Do something*/} ) 

Aún así, no hay dados. Estoy completamente perplejo; ¿que debería hacer? ¿Dónde debería poner esa llamada para execute() para que se ejecute en un hilo diferente? ¿Me estoy perdiendo de algo?

Vamos a romper esto un poco.

 Single.just(service?.people() ?.get("people/me") ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers") ?.execute() ?: Person()) .subscribeOn(Schedulers.io()) 

Estamos empezando con Single.just(someValue) . Podríamos reescribir su código como tal, y sería técnicamente idéntico a lo que escribió:

 val person = service?.people() ?.get("people/me") ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers") ?.execute() ?: Person()) Single.just(person) 

¿Ves el problema? La llamada de service se ejecuta primero para proporcionar un valor a la llamada Single.just .

Para ejecutar su llamada de service una vez que el rest de la cadena haya sido configurada (es decir, al momento de la suscripción), puede usar el defer :

 Single.defer(() -> service?.people() ?.get("people/me") ?.setPersonFields("emailAddresses,birthdays,genders,phoneNumbers") ?.execute() ?: Person()) .subscribeOn(Schedulers.io()) 

(la syntax puede estar desactivada, no tengo un IDE a mano)

  • ¿Alguna diferencia entre "Subject.asObservable ()" y el tema en sí "Subject"?
  • RxJava Observable.create envolver suscripciones observables
  • Cómo upload el valor de retorno de un constructor
  • Excepción: blockingConnect no se debe invocar en el subprocess UI a pesar de que subsubscribí en otro subprocess
  • Rxjava retrofit parse api error para el usuario
  • Comportamiento con Kotlin Higher-Order Functions e interfaces de método único?
  • Extender RxJava Observable en Kotlin con eliminación adecuada
  • RXJava Ignorar error y continuar en cadena
  • ¿Cómo puedo pausar que un evento fluya a través de un observable?
  • Cómo recordar el estado con los operadores de rebashs en RxJava2
  • RxAndroid, cómo detectar si observable ha finalizado la emisión