RxJava se usa en una gran cantidad de aplicaciones de Android, pero al mismo tiempo muchas no conocen otras fuentes de eventos, excepto Observable y quizás Flowable. Se olvidan de las clases especializadas Individual, Quizás y Completable, que a menudo pueden agregar más claridad al código.
Debajo del gato encontrará una hoja de trucos sobre las fuentes de eventos que existen en RxJava.
Completable es en realidad un análogo Rx de Runnable. Es una operación que se puede realizar o no. Si dibujamos una analogía con Kotlin, entonces esto es
divertido completable () del mundo de Rx. En consecuencia, para suscribirse, debe implementar onComplete y onError. No se puede crear a partir del valor (Observable # solo, ...) porque no está diseñado para esto.
Llamable reactiva simple, porque aquí es posible devolver el resultado de la operación. Continuando la comparación con Kotlin, podemos decir que Single es divertido single (): T {}. Por lo tanto, para suscribirse, debe implementar onSuccess (T) y onError.
Quizás : un cruce entre Single y Completable, porque admite un valor, sin valores y error. Es más difícil trazar un paralelo inequívoco con los métodos, pero creo que Quizás es divertido, quizás (): T? {}, que devuelve nulo cuando no hay resultado. Es fácil adivinar que para una suscripción debe definir onSuccess (T), onComplete y onError.
Es importante tener en cuenta que onSuccess (T) y onComplete son mutuamente excluyentes. Es decir en caso de llamar al primero, no puede esperar al segundo.
Observable es la fuente más común, debido a su versatilidad. Él sabe cómo no producir eventos y generar muchos de ellos, por lo que puede usarse siempre que otras opciones no sean adecuadas. A pesar de esto, Observable tiene un inconveniente: no sabe cómo manejar la contrapresión. Para suscribirse, necesita onNext (T), onError y onComplete.
Contrapresión: una situación en la que los nuevos eventos llegan mucho más rápido de lo que tienen tiempo para procesar y comienzan a acumularse en el búfer, desbordándolo. Esto puede generar problemas como OutOfMemoryError. Más detalles se pueden encontrar aquí .
ConnectableObservable : una versión calentada de Observable. Todas las fuentes de datos comienzan a emitir su flujo de eventos al momento de la suscripción. Pero no este chico. Para hacer esto, ConnectableObservable espera una llamada para conectarse. Esto se hace para que varios observadores puedan revisar una secuencia de eventos sin reiniciarla en cada suscripción. Para ilustrar, le daré el siguiente fragmento:
val observable = Observable.fromCallable { Log.d("RxLogs", "observable fromCallable executed") Thread.sleep(1000) }.subscribeOn(Schedulers.computation()) observable.subscribe() observable.subscribe() observable.subscribe() observable.subscribe()
En la consola estará:
observable desde Callable ejecutado
observable desde Callable ejecutado
observable desde Callable ejecutado
observable desde Callable ejecutado
val connectedObservable = Observable.fromCallable { Log.d("RxLogs", "connectedObservable fromCallable executed") Thread.sleep(1000) }.subscribeOn(Schedulers.computation()) .publish() connectedObservable.subscribe() connectedObservable.subscribe() connectedObservable.subscribe() connectedObservable.subscribe() connectedObservable.connect()
Y en este caso: observable desde Callable ejecutado
Fluible : una fuente que proporciona operadores adicionales para procesar la contrapresión. Cuando necesite manejar más de 10,000 eventos que ocurren rápidamente uno tras otro, se recomienda usarlo en lugar de Observable.
Este último puede crear un ConnectableFlowable, abriendo las mismas posibilidades que ConnectableObservable.
Hablando de generadores de eventos, uno no puede dejar de mencionar Asunto y Procesador.
Asunto : una clase que puede ser tanto una fuente como un navegador. Esto le permite usarlo, por ejemplo, en varios tipos de controladores que lo enviarán al exterior como un Observable y dentro lo notificarán como un Observador. A continuación, veremos diferentes implementaciones de esta clase.
AsyncSubject / AsyncProcessor retiene el último evento hasta que el subproceso finaliza correctamente y luego lo envía a los suscriptores. Si se produce un error, no se reenviarán eventos.
PublishSubject / PublishProcessor reenvía los eventos que entran más allá hasta que llega una señal de terminal. Después del final de la secuencia o error, devuelve los eventos apropiados.
BehaviorSubject / BehaviorProcessor funciona de manera similar a PublishSubject / PublishProcessor, pero al suscribirse devuelve el último evento, si lo hay, y si el Asunto no ha pasado al estado terminal.
ReplaySubject / ReplayProcessor - BehaviourSubject / BehaviorProcessor en esteroides. No devuelve un último evento, sino todo lo que el alma desea. Si se suscribe a un ReplaySubject o ReplayProcessor completado, se recibirán todos los datos acumulados.
Por lo tanto, ReplaySubject.createWithSize (1) y BehaviourSubject.create () funcionan de manera diferente después de la transición al estado terminal. Durante la suscripción, el primero devolverá el último evento y el segundo no. También es cierto para ReplayProcessor.
CompletableSubject ,
MaybeSubject y
SingleSubject funcionan de manera similar a PublishSubject, solo diseñado para usarse con Completable, Maybe y Single, respectivamente.
UnicastSubject / UnicastProcessor es en realidad un ReplaySubject que garantiza que solo tiene un suscriptor. Lanza una IllegalStateException cuando intenta volver a suscribirse.
Es decir siguiente fragmento
val subject = UnicastSubject.create<String>(3) repeat(3) { subject.onNext(it.toString()) } subject.onComplete() subject.subscribe({ Log.d("RxLogs", it) }, { }, { Log.d("RxLogs", "complete") })
saldrá al registro
0 0
1
2
completar
MulticastProcessor funciona de manera similar a PublishProcessor, con la excepción de una pequeña característica. Él sabe cómo manejar la contrapresión para la transmisión entrante. MulticastProcessor le permite establecer el tamaño del búfer en el que realizará una preconsulta de elementos ascendentes para futuros suscriptores.
En el siguiente diagrama, se crea un procesador con almacenamiento para 2 elementos, que solicita inmediatamente de su fuente. Por lo tanto, cuando el primer observador se suscribe, emite inmediatamente el contenido del búfer, que se llena instantáneamente con nuevos eventos. Después del evento terminal, MulticastProcessor borra su almacenamiento y los nuevos suscriptores reciben de inmediato la finalización de la transmisión.
