Articles of rx java

¿Hay alguna manera de cambiar mi método a la stream Observable que será una cadena de modificadores?

Originalmente tengo una function como esta: private fun rxSaveEvents( events: List<LocationEvent>, locationUpdatesAboutToStop: Boolean): Boolean { synchronized(mEventsMonitor) { if (events.isNotEmpty()) { val locationEvents = LocationEvents() segregateLocationEvents(events, locationEvents) scoreReadings(locationEvents) sendLocationEventsInternal(locationEvents, locationUpdatesAboutToStop) Completable.fromCallable { updateLocationEventsToDb(locationEvents) }.subscribeOn(Schedulers.io()).subscribe() } if (locationUpdatesAboutToStop) { stopLocationUpdates() Completable.fromCallable { cleanAllReadingsData() }.subscribeOn(Schedulers.io()).subscribe() Completable.fromCallable { removeExpinetworkingData() }.subscribeOn(Schedulers.io()).subscribe() } locationEvents.clear() return true } } Esta function se […]

Observable.fromCallable () implementación con exception

Estoy tratando de ejecutar una function que arroja excepciones. Observable.fromCallable { foo() } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .take(1) .onErrorReturn { /* onError operation */ } .subscribe { } y quiero hacer alguna otra operación si tiene éxito sin ningún error. ¿Cómo puedo hacer eso?

Cómo llenar una vista de list usando Kotlin, retroadaptación y RXjava

Mi integración con el service está list, puedo ver mi matriz en el logging de requestes, pero no puedo completar la list. Honestamente ya he intentado todo lo que sé, soy nuevo con kotlin, el lenguaje me hace un poco confundido. Alguien podría identificar dónde podría estar el error, lo que estoy haciendo mal. json […]

¿Por qué tengo un Log de salida no deseado al fusionar 2 observables en otro Observable, que los almacena temporalmente cada 10 segundos?

Estoy tratando de simular la fusión de 2 secuencias observables diferentes, que están emitiendo algún object por segundo. Este object tiene el mismo padre en esas 2 secuencias. Pensé, que en la console aparecerá un nuevo object con valor después de 1 segundo. Sin embargo, cuando imprimo esos objects, obtengo objects que saltearon la emisión […]

OnComplete nunca se llamó con toSortedList () y groupBy ()

Actualmente estoy usando la Biblioteca de ReactiveLocation de Android ( Github ). LastKnownLocationObservable ( Code ) funciona según lo previsto. Estoy usando flatMap para search estaciones cercanas desde un DB y (debido a su dominio) estoy creando un model a partir de los datos. Así que tengo una list de elementos y estoy creando el […]

MissingMethodInvocationException probando una class abierta en Kotlin

Estoy intentando hacer algunas testings de integración con Kotlin + RxJava y MockWebServer. Soy un poco novato en términos de testings y soy aprendiz de Kotlin. Conozco las limitaciones de Mockito y las classs finales, pero no debería burlarme de la class que trato de probar, así que no sé cuál es el verdadero problema: […]

Fusionar datos de diferentes Observables y elegir diferentes estrategias de búsqueda, según la disponibilidad de datos

Tengo una situación en la que, al search un object Item , debe ocurrir una de estas tres cosas: Si el elemento no está presente en la memory caching (también conocido como nulo), cargue datos de api1 y api2 , api2 los datos y devuelva un Observable . Si el elemento está presente en la […]

RxJava – ¿Entradas de keyboard de contrapresión?

Aquí hay un problema divertido de RxJava. Quiero usar los operadores de contrapresión de RxJava para search rápidamente una input mecanografiada mientras se escribe cada carácter, al igual que hace Google en su página de búsqueda. Revisé la documentation de Backpressure y se me ocurrió esto (estoy usando RxJavaFX / RxKotlinFX para aprovechar JavaFX). val […]

Rx Java Retrofit con flatMap se ejecuta solo una vez

Estoy tratando de get un observable de una secuencia de button, click flatmap con otro observable de la modificación para solicitar un punto final, ¡pero solo se ejecutará una vez! Cuando hago clic de nuevo, no paso por el método de map plano. Es extraño porque cuando el map plano devuelve otros observables, funciona bien, […]

Datos de caching de request HTTP en RxJava

Tengo esta FilmStore que busca una película de una API REST. class ApiFilmsStore(private val tmdApi: TmdApi, private val converter: ApiFilmToFilmConverter) : FilmsStore { override fun get(filmId: String): Observable<Film> { return tmdApi.filmById(filmId).flatMap { apiFilm -> Observable.just(converter.convert(apiFilm)) } .subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()) } } Tengo este otro que es un decorador para devolver datos de la memory caching si existe: […]