7. La biblioteca RxJava
La biblioteca RxJava se basa en el siguiente concepto: un flujo de elementos de tipo T Observable<T> es observado por uno o más suscriptores (suscriptores, observadores, consumidores) Subscriber<T>. La biblioteca RxJava permite que el flujo Observable<T> se ejecute en un hilo T1 y su observador Subscriber<T> en un hilo T2 sin que el desarrolladortenga que preocuparse por gestionar el ciclo de vida de estos hilos ni por problemas naturalmente complejos, como el intercambio de datos entre hilos y su sincronización para ejecutar una tarea global. Por lo tanto, facilita la programación asíncrona.
Un flujo Observable<T> produce elementos de tipo T, que se pueden observar a medida que se generan. Si el observador y el observable (término que, por uso coloquial, designa al tipo Observable<T>) se encuentran en el mismo hilo, entonces el observable solo puede producir el elemento (i+1) cuando el observador haya consumido el elemento i. Hay pocos casos en los que esta arquitectura resulte útil. Si el observador y el observable no se encuentran en el mismo hilo, entonces el observable y su observador tienen comportamientos autónomos: el observable produce a su propio ritmo y el observador consume a su propio ritmo. Ahí radica el interés de la biblioteca. Hasta ahora siempre hemos hablado de un observador. En realidad, un observable puede tener cualquier número de observadores.
La biblioteca RxJava se adapta particularmente bien a la arquitectura vista en el párrafo 2 de la introducción y que recordamos aquí:
- en [1], una capa de servicio proporciona servicios, algunos de los cuales tardan en obtenerse (por ejemplo, solicitudes de red);
- esta capa de servicios es invocada por una interfaz gráfica [1] (Swing, Android, JavaFx). Si la capa de servicios se ejecuta en el mismo hilo que el método [swing] que la utiliza, la interfaz gráfica se congela (no responde) mientras espera el resultado del servicio;
- en [2], una delgada capa de adaptación implementada con RxJava permite presentar a la capa gráfica una implementación asíncrona del mismo servicio: este puede ejecutarse en un hilo diferente al del método de la capa gráfica que lo invoca. En este caso, la interfaz gráfica [3] sigue siendo interactiva: el usuario puede seguir interactuando con ella, por ejemplo, iniciar una nueva solicitud de red en paralelo a la primera y, sobre todo, se le puede ofrecer la posibilidad de cancelar procesos que tardan demasiado, algo imposible si la interfaz gráfica está congelada;
- la llamada [4] es síncrona, mientras que la llamada [5-6] es asíncrona;
En esta arquitectura, la capa [2] ofrece servicios que devuelven tipos Observable<T> a los que pueden suscribirse los métodos de la capa gráfica [3]. Un servicio de la capa [2] entrega entonces sus resultados uno por uno, y la capa [3] puede reaccionar ante cada uno de ellos, por ejemplo, actualizando uno o varios componentes de la interfaz gráfica.
La clase Observable<T> cuenta con varias decenas de métodos. Esta es una de las dificultades de la biblioteca: es muy completa y resulta difícil comprender todas sus posibilidades. Presentaremos algunas de ellas. El dominio de los demás métodos vendrá con el tiempo.
7.1. Crear observables y suscribirse a ellos
7.1.1. Ejemplo-01: el método [Observable.from]
 |
Consideremos el siguiente código:
package dvp.rxjava.observables;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import java.util.Arrays;
public class Exemple01 {
public static void main(String[] args) {
// observables de enteros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
obs1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.printf("next : %s%n", integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println(throwable);
}
}, new Action0() {
@Override
public void call() {
System.out.println("completed");
}
});
}
}
- línea 12: se crea un tipo Observable<Integer> a partir de una lista de enteros.
La clase Observable<T> es un flujo de elementos de tipo T que se pueden observar, preferiblemente de manera asíncrona, aunque no necesariamente, a medida que se producen. Su definición es la siguiente:
 |
Como ya se ha mencionado, la clase Observable<T> cuenta con varias decenas de métodos. Algunos son similares a los de la clase Stream<T> analizada en el párrafo 5. La documentación de RxJava incluye «diagramas de mármol» [2] que ilustran el funcionamiento de estos métodos:
- la línea 3 ilustra las emisiones del observable a lo largo del tiempo;
- el método [4] se aplica a los elementos emitidos por el observable. Por lo general, produce un nuevo observable;
- la línea 5 muestra el nuevo observable obtenido;
El método [Observable.from] tiene la siguiente firma:
 |
El método estático [Observable.from] permite crear un Observable<T> a partir de una colección de elementos de tipo T. Es una forma muy sencilla de comenzar con los observables. La línea:
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
emitirá, por lo tanto, tres elementos. No los emite de inmediato. Los emitirá en su totalidad cada vez que se registre un suscriptor. A esto se le llama un observable frío. El observable vuelve a emitir sus elementos por cada nuevo suscriptor.
Podemos considerar la instrucción anterior como una acción de configuración del observable. Este se configura una vez y se ejecuta n veces si se presentan n suscriptores.
¿Cómo se suscribe uno?
Una forma de hacerlo es utilizar el método [Observable.subscribe], cuya definición utilizada aquí es la siguiente:
 |
- el primer parámetro [Action1<T> onNext] (véase el párrafo 6.2) del método es el método que se debe ejecutar cuando el observable emite un nuevo elemento T;
- el segundo parámetro [Action1<Throwable> onError] del método es el método que se debe ejecutar cuando el observable genera una excepción;
- el tercer parámetro [Action0 onComplete] (véase el párrafo 6.1) del método es el método que se debe ejecutar cuando el observable lanza una excepción;
- el método devuelve un tipo [Subscription];
El tipo [Subscription] representa una suscripción al observable. Su definición es la siguiente:
 |
El interés de esta interfaz [1] radica en su método [2], que permite cancelar una suscripción.
En nuestro ejemplo, el código de la suscripción al observable es el siguiente:
obs1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.printf("next : %s%n", integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println(throwable);
}
}, new Action0() {
@Override
public void call() {
System.out.println("completed");
}
});
- línea 1: se ignora el resultado de tipo [Subscription];
- líneas 1-15: los tres parámetros son instancias de clases anónimas. También utilizaremos lambdas. La ventaja de las clases anónimas es que se ven claramente los tipos de datos que espera el único método de estas clases;
- líneas 2-5: implementación del primer parámetro de tipo [Action1<Integer>];
- líneas 6-10: implementación del segundo parámetro de tipo [Action1<Throwable>];
- líneas 11-15: implementación del tercer parámetro de tipo [Action0];
El código completo es el siguiente:
package dvp.rxjava.observables;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import java.util.Arrays;
public class Exemple01 {
public static void main(String[] args) {
// observables de enteros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// suscripción
obs1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.printf("next : %s%n", integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println(throwable);
}
}, new Action0() {
@Override
public void call() {
System.out.println("completed");
}
});
}
}
El observable de la línea 12 comienza a emitir sus tres elementos tan pronto como se invoca el método [subscribe] en la línea 14. A partir de ese momento:
- con cada elemento emitido, se ejecutan las líneas 15 a 18.
- al finalizar los tres elementos, se ejecutan las líneas 24 a 29;
- las líneas 19 a 24 nunca se ejecutarán porque el observable no emite una excepción aquí;
Por defecto, el observable y el observador se ejecutan en el mismo hilo. Existen algunos observables predefinidos que se ejecutan en un hilo diferente al hilo principal (en este caso, el hilo del método main), pero para la mayoría de ellos no es así. Por lo tanto, aquí todo ocurre en el hilo del método [main]:
- el observable emite el elemento 1;
- las líneas 15-18 se ejecutan y muestran ese elemento;
- el observable emite el elemento 2;
- se ejecutan las líneas 15 a 18 y muestran este elemento;
- el observable emite el elemento 3;
- se ejecutan las líneas 15 a 18 y muestran este elemento;
- el observable emite la notificación [completed];
- se ejecutan las líneas 24 a 29;
Esto es lo que muestran los resultados obtenidos:
La clase [Exemple02] retoma [Exemple01], utilizando esta vez funciones lambda como parámetros del método [Observable.subscribe]:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
public class Exemple02 {
public static void main(String[] args) {
// observables de números enteros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// suscripción
obs1.subscribe(
(integer) -> System.out.printf("next : %s%n", integer),
(th) -> System.out.println(th),
() -> System.out.println("completed"));
}
}
7.1.2. Ejemplo-03: la clase Observer
 |
El método [Observable.subscribe], que permite suscribirse a un observable, tiene varias versiones, entre ellas la siguiente:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
import rx.Observer;
public class Exemple03 {
public static void main(String[] args) {
// observables de números enteros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// suscripción
obs1.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("completed");
}
@Override
public void onError(Throwable th) {
System.out.printf("throwable %s", th);
}
@Override
public void onNext(Integer integer) {
System.out.printf("next : %s%n", integer);
}
});
};
}
En la línea 13, en lugar de pasar tres parámetros al método [subscribe], se le pasa un tipo [Observer] como se muestra a continuación:
 |
El tipo [Observer] es una interfaz con tres métodos:
- [onNext(T t)], que se invoca cada vez que el observable emite un elemento t;
- [onError(Throwable th)], que se invoca cuando el observable lanza una excepción th;
- [onCompleted], que se invoca cuando el observable indica que ha terminado de emitir;
El funcionamiento del código es similar al explicado anteriormente. Se obtienen los siguientes resultados:
7.1.3. Ejemplo-04: el método [Observable.create]
 |
El método estático Observable.create se define de la siguiente manera:
 |
- el método [create] devuelve un tipo Observable<T>;
- el parámetro del método [create] es una función de tipo [Observable.OnSubscribe<T>] definida de la siguiente manera:
 |
El tipo [Observable.OnSubscribe<T>] es una interfaz funcional que, a su vez, extiende la interfaz funcional [Action1<Subscriber<? super T>>]. El método [call] de esta interfaz espera un tipo [Subscriber] (suscriptor, observador) definido de la siguiente manera:
 |
En [1], se observa que la clase [Subscriber<T>] implementa la interfaz [Observer<T>] presentada en el párrafo 7.1.2.
Finalmente, el método [<T> Observable.create]:
- espera como parámetro una instancia de tipo [Observable.OnSubscribe<T>] que tenga un único método con la siguiente firma: void call(Subscriber<T> s). El tipo [Subscriber<T>] extiende el tipo [Observer<T>] y, por lo tanto, cuenta con los métodos onNext, onError y onCompleted;
- devuelve un tipo Observable<T>;
El método [<T> Observable.create] devuelve un observable configurado. Aún no se ha emitido ningún elemento. Cuando un suscriptor [Subscriber<T> s] se suscribe a este observable, se invoca el método [void call(s)] de la función pasada como parámetro del método [<T> Observable.create]. Su función es emitir elementos t de tipo T y llamar al método [s.onNext(t)] del observador en cada emisión. Cuando este haya finalizado, se debe llamar al método [s.onCompleted(t)] del observador y el método [call] debe finalizar. Si el método [call] encuentra una excepción th, se debe llamar al método [s.onError(th)] del observador y el método [call] debe finalizar;
Para ilustrar este complejo funcionamiento, utilizaremos el siguiente código [Exemple04]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import java.util.Random;
public class Exemple04 {
public static void main(String[] args) {
// configuración observable de números reales
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
for (int i = 0; i < 3; i++) {
// emisión del elemento i
subscriber.onNext(new Random((i + 1)).nextDouble());
}
// fin de emisión
subscriber.onCompleted();
}
});
// suscripción y, por lo tanto, emisión
obs1.subscribe((d) -> System.out.printf("onNext %s%n", d), (th) -> System.out.printf("onError %s%n", th),
() -> System.out.println("onCompleted"));
}
}
- línea 11: se crea un observable que emite tipos Double;
- líneas 11-21: el parámetro del método [create] se instancia con una clase anónima que tiene el único método [call] de las líneas 12-20. El observable creado en la línea 11 está listo para emitir, pero solo lo hará cuando llegue un observador;
- líneas 13-21: el método [call] recibe la referencia de un observador;
- líneas 14-17: emisión de 3 elementos hacia el observador;
- línea 19: notificación de fin de transmisión al observador;
- líneas 23-24: suscripción al observable de la línea 11. Se implementan los tres parámetros [onNext, onError, onCompleted] del método [subscribe] mediante tres lambdas. Esta suscripción creará el suscriptor [Subscriber<Double>], el cual se pasará al método [call] de la línea 13. A continuación, comenzará la emisión de elementos;
- todo ocurre en el mismo hilo: observable y observador;
Se obtienen los siguientes resultados:
El método [Observable.create] permite crear un observable a partir de cualquier fenómeno. Este es el método que utilizamos en el párrafo 2 de la sección de descubrimiento, para transformar una interfaz síncrona en una asíncrona.
7.1.4. Ejemplo-05: refactorización de [Exemple-04]
 |
El siguiente ejemplo presenta una nueva versión del método estático [Observable.subscribe]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
public class Exemple05 {
public static void main(String[] args) {
// configuración de un observable de números reales
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
showInfos("Observable.call start");
for (int i = 0; i < 3; i++) {
// espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// error
subscriber.onError(e);
}
// acción
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// finalizado
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// un suscriptor
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
}
@Override
public void onError(Throwable e) {
showInfos(String.format("Subscriber.onError (%s)", e));
}
@Override
public void onNext(Double aDouble) {
showInfos(String.format("Subscriber.onNext (%s)", aDouble));
}
};
// suscripción
showInfos("avant souscription");
obs1.subscribe(subscriber);
showInfos("après souscription");
}
private static void showInfos(String message) {
System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message, Thread.currentThread().getName(),
new SimpleDateFormat("ss:SSS").format(new Date()));
}
}
- línea 56: la nueva versión del método estático [Observable.subscribe] admite como parámetro el tipo [Subscriber] que presentamos en el párrafo anterior;
- líneas 37-52: el suscriptor (suscriptor, observador). Implementa la interfaz Observer con sus tres métodos onNext, onError y onCompleted;
- líneas 61-64: a partir de ahora nos centraremos en los hilos en los que se ejecutan el observable y su observador;
- línea 62: el nombre del hilo;
- línea 63: la hora actual expresada en segundos y milisegundos. Esto nos permitirá ver a lo largo del tiempo la emisión de elementos por parte del observable y su procesamiento por parte del observador;
- Este código tiene la misma funcionalidad que el código anterior. Simplemente hemos refactorizado este último;
Los resultados obtenidos son los siguientes:
- línea 1 de los resultados: antes de la línea 56 del código, aún no ha pasado nada. El observable simplemente se ha configurado;
- línea 2 de los resultados: la línea 56 del código provoca la llamada al método [call] de la línea 15. En la línea 3, se envía el número real 80.39 al observador;
- línea 4: el observador recibe el número enviado;
- líneas 5-8: el proceso anterior se repite dos veces;
- línea 9: el observable envía la notificación de fin de transmisión;
- línea 10: el observador la recibe;
- línea 11: mostrada por la línea 57 del código;
Así pues, vemos que la única línea 56 de suscripción provocó que se mostraran las líneas 2 a 10 de los resultados. Cuando se empieza a trabajar con la biblioteca RxJava, uno se pregunta cómo se encadenan las cosas entre sí y, en particular, cuáles son los vínculos que unen al observador y al observable. Aquí vemos que la línea 56, la suscripción al observable,
- ha provocado la emisión de todos los elementos del observable;
- que el observable y el observador se ejecutan en el mismo hilo;
- que, debido a ello, se observa la secuencia: emisión del elemento i, observación del elemento i, emisión del elemento (i+1), observación del elemento (i+1), ...
Recordemos que el emisor esperaba antes de emitir sus elementos:
// en espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// error
subscriber.onError(e);
}
donde i en la línea 3 representa el número de emisión (0<=i<3). Si observamos las horas de emisión de los elementos del observable:
- líneas 2 y 3: el elemento 0 se transmitió aproximadamente 500 ms después del inicio de la suscripción;
- líneas 3 y 5: el elemento 1 se transmitió aproximadamente 400 ms después del elemento 0;
- líneas 5 y 7: el elemento 2 se emitió aproximadamente 300 ms después del elemento 1;
7.2. Hilo de ejecución, hilo de observación
7.2.1. Ejemplo-06: observable y observador en un hilo distinto de [main]
 |
Reestructuramos el ejemplo anterior de la siguiente manera [Exemple06]:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
public class Exemple06 {
public static void main(String[] args) {
// guardia de barrera
CountDownLatch latch = new CountDownLatch(1);
// configuración de un observable de números reales
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
showInfos("Observable.call start");
for (int i = 0; i < 3; i++) {
// en espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// error
subscriber.onError(e);
}
// acción
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// finalizado
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// un suscriptor
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// se baja la barrera
latch.countDown();
}
@Override
public void onError(Throwable e) {
showInfos(String.format("Subscriber.onError (%s)", e));
}
@Override
public void onNext(Double aDouble) {
showInfos(String.format("Subscriber.onNext (%s)", aDouble));
}
};
// continuación de la configuración observable
obs1 = obs1.subscribeOn(Schedulers.computation());
// suscripción
showInfos("avant souscription");
obs1.subscribe(subscriber);
// espera frente a la barrera
try {
showInfos("début attente barrière");
latch.await();
showInfos("fin attente barrière");
} catch (InterruptedException e1) {
System.out.println(e1);
}
showInfos("après souscription");
}
private static void showInfos(String message) {
System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message, Thread.currentThread().getName(),
new SimpleDateFormat("ss:SSS").format(new Date()));
}
}
- línea 16: creamos una barrera (semáforo) con un objeto de tipo [CountDownLatch]. Este objeto sirve para sincronizar hilos entre sí. Aquí se inicializa con el valor 1, al que llamaremos el valor de la barrera (o del semáforo). Un hilo se pone en espera de la barrera mediante una operación:
latch.await();
El hilo queda bloqueado si el valor del guardabarriera es >0. Un hilo puede aumentar o disminuir el valor interno del guardabarriera. En la línea 48, el valor del guardabarriera se decrementa en 1.
- Línea 63: el observable se configura para ejecutarse en un hilo proporcionado por el programador [Schedulers.computation()]. Este programador puede proporcionar tantos hilos como núcleos haya en la máquina de ejecución. El párrafo sobre la aplicación de ejemplo mostró el uso de otros programadores (véase el párrafo 2.8);
El principio del código es el siguiente:
- el método [main] se ejecuta en el hilo principal (main);
- línea 66: inicia la emisión de elementos del observable. Estos se emitirán en un hilo distinto al hilo principal;
- línea 70: el hilo principal queda bloqueado porque el valor del guardabarriera es 1 (véase la línea 16). Solo podrá continuar cuando este valor cambie a 0. Esto ocurre en la línea 48. Es el observador quien baja la barrera cuando recibe la notificación de que el observable ha terminado sus emisiones;
La ejecución arroja los siguientes resultados:
- línea 1: se llevará a cabo la suscripción;
- línea 2: esto desencadena la ejecución del método [call] en el hilo [RxComputationThreadPool-1]. Ahora tenemos una ejecución paralela con dos hilos;
- línea 3: por una razón no aclarada, el hilo [RxComputationThreadPool-1] cedió el control. El hilo [main] toma entonces el control y queda bloqueado por el guardabarriera (línea 70 del código). A partir de ese momento, solo el hilo [RxComputationThreadPool-1] puede operar;
- líneas 4-11: se observa el comportamiento visto anteriormente entre el observable y su observador, pero ahora todo ocurre en el hilo [RxComputationThreadPool-1];
- líneas 12-13: el observador ha bajado la barrera (línea 48 del código) y el hilo [RxComputationThreadPool-1] ha finalizado. El hilo [main] toma el control y muestra dos mensajes;
7.2.2. Ejemplo-07: observable y observador en dos hilos diferentes
|
Modificamos el ejemplo anterior de la siguiente manera:
package dvp.rxjava.observables;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
public class Exemple07 {
public static void main(String[] args) {
// guardia de la barrera
CountDownLatch latch = new CountDownLatch(1);
// configuración de un observable de números reales
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
showInfos("Observable.call start");
for (int i = 0; i < 3; i++) {
// espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// error
subscriber.onError(e);
}
// acción
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// finalizado
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// un suscriptor
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// se baja la barrera
latch.countDown();
}
@Override
public void onError(Throwable e) {
showInfos(String.format("Subscriber.onError (%s)", e));
}
@Override
public void onNext(Double aDouble) {
showInfos(String.format("Subscriber.onNext (%s)", aDouble));
}
};
// continuación de la configuración observable
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
// suscripción
showInfos("avant souscription");
obs1.subscribe(subscriber);
// esperando que suba la barrera
try {
showInfos("début attente barrière");
latch.await();
showInfos("fin attente barrière");
} catch (InterruptedException e1) {
System.out.println(e1);
}
showInfos("après souscription");
}
private static void showInfos(String message) {
System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message, Thread.currentThread().getName(),
new SimpleDateFormat("ss:SSS").format(new Date()));
}
}
El código es idéntico al del ejemplo anterior, excepto por la línea 63:
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
que configura el observable (subscribeOn) y el observador (observeOn) para que se ejecuten en uno de los hilos proporcionados por el programador [Schedulers.computation()].
Los resultados obtenidos son los siguientes:
Se pueden observar los siguientes puntos:
- el observable se ejecuta en el hilo [RxComputationThreadPool-4] (líneas 3-4, 6, 8-9);
- el observador se ejecuta en el hilo [RxComputationThreadPool-3] (líneas 5, 7, 10-11);
- que se ejecutan de manera independiente. Así, en las líneas 8-9, el observable emite dos notificaciones (onNext, onCompleted) antes de que el observador recoja la notificación [onNext] (línea 10);
La biblioteca RxJava se encarga de la transferencia de datos (las emisiones) del hilo del observable al hilo del observador. El desarrollador no tiene que preocuparse por ello.
Hemos visto cómo crear observables (Observable.from, Observable.create). Ahora veremos los observables predefinidos de la biblioteca RxJava.
7.3. Observables predefinidos
7.3.1. Ejemplo-08: el método [Observable.range]
 |  |
A partir de ahora, utilizaremos clases específicas para los procesos observados y sus observadores. La idea es poder registrar su nombre, su hilo de ejecución y las horas de ejecución para poder dar seguimiento a estas a lo largo del tiempo.
La clase [Process] será simplemente un Observable al que se le puede asignar un nombre. Implementará la siguiente interfaz [IProcess]:
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcess<T> {
// nombre de la variable observable
public String getName();
// observable
public Observable<T> getObservable();
}
Esta interfaz podrá ser implementada por la siguiente clase [Process<T>]:
package dvp.rxjava.observables.utils;
import rx.Observable;
import rx.Scheduler;
public class Process<T> implements IProcess<T>{
// nombre de la variable observable
protected String name;
// proceso observado
protected Observable<T> observable;
// constructores
public Process(String name, Observable<T> observable) {
// inicializaciones locales
this.name = name;
this.observable = observable;
}
// getters y setters
public String getName() {
return name;
}
public Observable<T> getObservable() {
return observable;
}
}
- línea 9: el nombre del proceso;
- línea 11: la variable observada;
- líneas 14-18: el constructor;
El observador se describirá mediante la siguiente clase [Observateur]:
package dvp.rxjava.observables.utils;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import rx.Subscriber;
public class Observateur<T> extends Subscriber<T> {
...
}
- línea 11, la clase Observateur<T> extiende la clase Subscriber<T> que presentamos brevemente en el párrafo 7.1.3. La utilizaremos como argumento del método [Observable.subscribe]:
// ejecución observable (observación)
obs1.subscribe(observateur);
El método [Observable.subscribe] utilizado en la línea 2 anterior tiene la siguiente definición:
 |
La función del [Subscriber] es principalmente gestionar los elementos emitidos por el observable al que se ha suscrito mediante los métodos de la interfaz [Observer]: onNext, onError, onCompleted. La clase [Subscriber] cuenta con los siguientes métodos:
 |
En el código de la clase [Observateur], utilizaremos el método [1] isUnsubscribed para saber si la suscripción del suscriptor se ha cancelado o no. La clase [Observateur<T>] completa es la siguiente:
package dvp.rxjava.observables.utils;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import rx.Subscriber;
public class Observateur<T> extends Subscriber<T> {
// un semáforo
private CountDownLatch latch;
// un método de visualización
private Consumer<String> showInfos;
// el nombre del observador
private String observerName;
// el nombre del proceso observado
private String processName;
// constructores
public Observateur() {
}
public Observateur(String name, CountDownLatch latch, Consumer<String> showInfos, String observedName) {
this.observerName = name;
this.latch = latch;
this.showInfos = showInfos;
this.processName = observedName;
}
// --------------------------- implementación de la interfaz Observer<T>
@Override
public void onCompleted() {
// fin de las transmisiones
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s,%s].onCompleted", observerName, processName));
}
// fin del bloqueo del hilo principal
latch.countDown();
}
@Override
public void onError(Throwable e) {
// error de emisión
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s, %s].onError (%s)", observerName, processName, e));
}
}
@Override
public void onNext(T value) {
// una emisión adicional
if (!isUnsubscribed()) {
try {
showInfos.accept(String.format("Subscriber [%s,%s] : onNext (%s)", observerName, processName,
new ObjectMapper().writeValueAsString(value)));
} catch (JsonProcessingException e) {
showInfos.accept(String.format("Subscriber [%s,%s].onNext (%s)", observerName, processName, e));
}
}
}
}
- Además de las características de un Subscriber, el observador Observateur incluirá la siguiente información:
- línea 14: un guardabarriera o semáforo que servirá para bloquear el hilo principal hasta que el observador haya recibido todos los elementos emitidos por el observable. Esto se hará en la línea 36 del código cuando el observador reciba del observable la notificación de fin de emisión;
- línea 16: una instancia de Consumer<String> que servirá para mostrar un mensaje en la consola;
- línea 18: el nombre del observador para distinguirlos entre sí cuando haya varios;
- línea 20: el nombre del proceso observado;
- líneas 36, 46, 54: los métodos [onCompleted, onError, onNext] de la interfaz [Observer<T>] implementada por la clase abstracta [Subscriber<T>]. Esta clase no los implementa. Por lo tanto, hay que hacerlo en sus clases hijas. Antes de realizar cualquier acción en estos métodos, se verifica si el observador no se ha dado de baja del observable que está observando;
- línea 59: el método [onNext] del observador escribe la cadena jSON del elemento recibido. Esto nos permitirá mostrar diversos tipos de elementos;
Dicho esto, analicemos un nuevo método de la clase Observable, el método [range]:
 |
El observable Observable.range(n,m) emite (m) enteros que van de n a n+m-1. Lo analizamos con el siguiente código [Exemple08]:
package dvp.rxjava.observables.exemples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import dvp.rxjava.observables.utils.Observateur;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple08 {
public static void main(String[] args) throws InterruptedException {
// número de observadores
final int nbObservateurs = 2;
// semáforo
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// configuración observable
Observable<Integer> obs1 = Observable.range(15, 3).subscribeOn(Schedulers.computation());
// ejecución observable (observación)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
obs1.subscribe(new Observateur<>(String.format("observateur[%d]", i), latch, showInfos,"obs1"));
}
// espera
showInfos.accept("main : attente fin observation");
latch.await();
// fin
showInfos.accept("main : fin observation");
}
// visualizaciones
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- línea 16: vamos a utilizar dos observadores;
- línea 19: el semáforo se inicializa en dos porque vamos a colocar a cada observador en un hilo diferente. Por lo tanto, el hilo principal deberá esperar a que finalicen los dos hilos de observación;
- línea 22: configuramos el observable de tal manera que se ejecute en un hilo del programador [Schedulers.computation()]. El observador estará en el mismo hilo que el observable;
- líneas 25-27: se suscriben dos observadores al observable. Esto provocará la ejecución completa de este para cada uno de los observadores: se emitirán los números enteros 15, 16 y 17;
- línea 30: el hilo principal espera a que terminen los observadores;
Los resultados obtenidos son los siguientes:
- línea 2: el hilo principal está bloqueado a la espera de que finalicen los dos observadores;
- líneas 3-4: se observa que el observador 0 está en el hilo [RxComputationThreadPool-1] y el observador 1 en el hilo [RxComputationThreadPool-2];
- líneas 3-10: se observa que ambos observadores reciben exactamente los mismos elementos;
Vamos a utilizar la clase Observateur así definida para ilustrar el comportamiento de otros tipos de observables.
7.3.2. Ejemplo-09: los métodos Observable.[interval, take, doNext]
 |
 |
Este ejemplo ilustra el uso del observable Observable.interval (intervalo largo, unidad TimeUnit), que emite enteros largos a intervalos de tiempo regulares. Cabe destacar el punto [1]: por defecto, el observable [Observable.interval] se ejecuta en uno de los hilos del programador [Schedulers.computation].
El código será el siguiente:
package dvp.rxjava.observables.exemples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import dvp.rxjava.observables.utils.Observateur;
import rx.Observable;
public class Exemple09 {
public static void main(String[] args) throws InterruptedException {
// número de observadores
final int nbObservateurs = 2;
// semáforo
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// configuración observable
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// ejecución observable (observación)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
obs1.subscribe(new Observateur<>(String.format("observateur [%d]", i), latch, showInfos,
"obs1"));
}
// espera
showInfos.accept("main : attente fin observation");
latch.await();
// fin
showInfos.accept("main : fin observation");
}
// visualizaciones
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- línea 22: el observable emite enteros largos cada 500 milisegundos. La serie comienza con el número 0;
- línea 22: este observable emite un número infinito de valores. El método [Observable.take(n)] crea un nuevo observable que conserva solo los primeros n elementos emitidos;
 |
Volvamos al código del observable:
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
En la línea 2, el método [Observable.doOnNext] se ejecuta cada vez que el observable emite un nuevo elemento. Esto se utiliza con frecuencia para registrar información. En este caso, queremos registrar la fecha de emisión de los elementos para verificar si se cumple el intervalo de 500 milisegundos. El método [Observable.doOnNext] no modifica el observable al que se aplica. Su definición es la siguiente:
 |
La ejecución arroja los siguientes resultados:
- líneas 3, 7 y 11: se observa que, aproximadamente, el intervalo de emisión es cercano a los 500 ms;
- Los dos observadores se encuentran, por supuesto, en dos subprocesos diferentes, a pesar de que el observable no se había configurado para ejecutarse con un programador específico. Este es el funcionamiento por defecto del observable [Observable.interval] que vemos aquí;
7.3.3. Ejemplos-10/12: los métodos Observable.[error, empty, never]
 |  |
A partir de ahora seremos más concisos en nuestras ilustraciones de los métodos de la clase [Observable]. El código anterior era el siguiente:
package dvp.rxjava.observables;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import rx.Observable;
public class Exemple09 {
public static void main(String[] args) throws InterruptedException {
// número de observadores
final int nbObservateurs = 2;
// semáforo
CountDownLatch latch = new CountDownLatch(nbObservateurs);
// configuración observable
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// ejecución observable (observación)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
obs1.subscribe(new Observateur<>(String.format("observateur [%d]", i), latch, showInfos,
"obs1"));
}
// espera
showInfos.accept("main : attente fin observation");
latch.await();
// fin
showInfos.accept("main : fin observation");
}
// visualizaciones
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
Este código ya se había utilizado en el ejemplo anterior. Solo cambiaban las líneas 21 y 22. Por lo tanto, vamos a factorizar la mayor parte de este código en la siguiente clase [ProcessUtils]:
package dvp.rxjava.observables.utils;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;
import rx.Observable;
public class ProcessUtils {
@SafeVarargs
public static void subscribe(int nbObservateurs, IProcess<?>... processes) throws InterruptedException {
// semáforo
CountDownLatch latch = new CountDownLatch(nbObservateurs * processes.length);
// ejecución observable (observación)
showInfos.accept("main : début observation");
for (int i = 0; i < nbObservateurs; i++) {
for (IProcess<?> process : processes) {
Observable<?> obs = process.getObservable();
obs.subscribe(new Observateur<>(String.format("observateur[%d]", i), latch, showInfos, process.getName()));
}
}
// espera
showInfos.accept("main : attente fin observation");
latch.await();
// fin
showInfos.accept("main : fin observation");
}
// visualizaciones
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- línea 13: el método admite dos parámetros:
- nbObservateurs: el número de observadores de los procesos pasados como segundo parámetro;
- processes: los procesos (observables con nombre) que se deben observar. Gracias a la notación [IProcess<?>], los procesos podrán emitir elementos de diferentes tipos;
- línea 16: el semáforo debe pasar a verde cuando todos los observadores hayan completado todas sus observaciones. El valor inicial del semáforo es, por lo tanto, el número de observadores multiplicado por el número de observaciones;
- líneas 20-25: se suscribe a cada observador a todos los procesos que debe observar;
- línea 23: se recupera el observable del proceso (véase el párrafo 7.3.1);
- línea 23: se suscribe un observador a la observable. Se le pasan cuatro datos:
- su nombre;
- el semáforo que debe decrementar cuando reciba la notificación de fin de emisión de la observable que está observando;
- el método que debe utilizar cuando quiera registrar información en la consola;
- el nombre del proceso que va a observar;
Una vez definidas estas clases, el ejemplo 10 será el siguiente:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple10 {
public static void main(String[] args) throws InterruptedException {
// configuración observable
Observable<?> obs = Observable.error(new RuntimeException("Erreur !!!")).subscribeOn(Schedulers.computation());
// ejecución (observación) observable
ProcessUtils.subscribe(2,new Process<>("process1", obs));
}
}
En la línea 11, el método estático [Observable.error] se define de la siguiente manera:
 |
La línea 8 configura, por lo tanto, un observable que simplemente lanza una excepción dirigida al método [onError] de sus suscriptores. La ejecución arroja los siguientes resultados:
main : début observation ------Thread[main] ---- Time[22:618]
main : attente fin observation ------Thread[main] ---- Time[22:636]
Subscriber[observateur[1], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-2] ---- Time[22:638]
Subscriber[observateur[0], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-1] ---- Time[22:638]
En las líneas 3 y 4, el método [onError] de ambos suscriptores recibió la excepción lanzada por el observable.
Esta ejecución tiene una particularidad: los métodos [onCompleted] de ambos observadores no se han invocado. Por lo tanto, la barrera no se ha bajado y el hilo principal permanece bloqueado en el método estático [ProcessUtils.subscribe] en la siguiente línea 3:
// espera
showInfos.accept("main : attente fin observation");
latch.await();
// fin
showInfos.accept("main : fin observation");
Aquí vemos que, en caso de error del observable, no se invoca el método [onCompleted] de los suscriptores. Por lo tanto, modificamos el método [Observateur.onError] de la siguiente manera:
@Override
public void onError(Throwable e) {
// error de transmisión
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber[%s, %s].onError (%s)", observerName, processName, e));
}
// fin del bloqueo del hilo principal
latch.countDown();
}
Agregamos las líneas 7 y 8 para eliminar la restricción en caso de error del observable. Con este nuevo código, la ejecución arroja los siguientes resultados:
main : début observation ------Thread[main] ---- Time[40:750]
main : attente fin observation ------Thread[main] ---- Time[40:764]
Subscriber[observateur[0], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-1] ---- Time[40:766]
Subscriber[observateur[1], process1].onError (java.lang.RuntimeException: Erreur !!!) ------Thread[RxComputationThreadPool-2] ---- Time[40:766]
main : fin observation ------Thread[main] ---- Time[40:767]
Obtenemos la línea 5, que no habíamos obtenido anteriormente.
El ejemplo 11 será el siguiente:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple11 {
public static void main(String[] args) throws InterruptedException {
// configuración observable
Observable<?> obs1 = Observable.empty();
// ejecución (observación) observable
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
En la línea 10, el método estático [Observable.empty] crea un observable que no emite ningún elemento. Solo emite la notificación de fin de emisión;
 |
La ejecución del código del ejemplo anterior arroja los siguientes resultados:
- líneas 2 y 3: se observa que ambos observadores reciben la notificación de fin de emisión sin haber recibido elementos previamente.
Cabe preguntarse para qué sirve este método. Se puede utilizar de manera análoga a una colección, vacía al inicio, en la que luego se van acumulando elementos:
En la línea 3, se fusiona el observable inicial obs (línea 1) con otros observables.
El ejemplo 12 ilustra el método estático [Observable.never]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple12 {
public static void main(String[] args) throws InterruptedException {
// configuración observable
Observable<?> obs1 = Observable.never();
// ejecución (observación) observable
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
El método estático [Observable.never] crea un observable que nunca emite:
 |
La ejecución del ejemplo arroja los siguientes resultados:
En la línea 2, el hilo principal espera indefinidamente. De hecho, ningún observable emite la notificación [onCompleted] que permite que el semáforo (barrera) cambie a verde (bajar la barrera).
7.4. Multi-threading
7.4.1. Ejemplo 13: hilo de acción, hilo de observación
En el párrafo 7.1.3 creamos un observable con el método estático [Observable.create]:
|
- el método [create] devuelve un tipo Observable<T>;
- el parámetro del método [create] es una función de tipo [Observable.OnSubscribe<T>] definida de la siguiente manera:
|
El tipo [Observable.OnSubscribe<T>] es una interfaz funcional que, a su vez, extiende la interfaz funcional [Action1<Subscriber<? super T>>]. El método [call] de esta interfaz espera un tipo [Subscriber] (suscriptor, observador). En el resto de este documento, en ocasiones nos referiremos al tipo [Observable.OnSubscribe<T>] como una acción. Vamos a crear acciones personalizadas que tendrán un nombre. Serán instancias de la siguiente interfaz [IProcessAction]:
 |
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcessAction<T> extends Observable.OnSubscribe<T> {
// la acción tiene un nombre
public String getName();
}
- línea 5: la interfaz [IProcessAction<T>] tiene todas las características de la interfaz [Observable.OnSubscribe<T>];
- línea 8: además, cuenta con un método [getName] que devuelve el nombre de la instancia que implementa la interfaz;
Vamos a utilizar la siguiente acción denominada [ProcessAction01]:
package dvp.rxjava.observables.utils;
import java.util.Random;
import rx.Subscriber;
import rx.functions.Func1;
public class ProcessAction01<T> implements IProcessAction<T> {
// datos
private String name;
private int nbValues;
private Func1<Integer, T> func1;
// constructores
public ProcessAction01(String name, int nbValues, Func1<Integer, T> func1) {
this.name = name;
this.nbValues = nbValues;
this.func1 = func1;
}
@Override
public void call(Subscriber<? super T> subscriber) {
ProcessUtils.showInfos.accept(String.format("Observable (%s) call start", getName()));
for (int i = 0; i < nbValues; i++) {
// espera
try {
Thread.sleep(new Random().nextInt(500));
} catch (InterruptedException e) {
// error
ProcessUtils.showInfos.accept(String.format("Observable (%s) onError", getName()));
subscriber.onError(e);
}
// emisión de un elemento
T value = func1.call(i);
ProcessUtils.showInfos.accept(String.format("Observable (%s,%s) onNext (%s)", getName(), i, value));
subscriber.onNext(value);
}
// finalizado
ProcessUtils.showInfos.accept(String.format("Observable (%s) onCompleted", getName()));
subscriber.onCompleted();
}
@Override
public String getName() {
return name;
}
}
- línea 8: la clase [ProcessAction01<T>] implementa la interfaz [IProcessAction<T>] y, por lo tanto, la interfaz [Observable.OnSubscribe<T>];
- línea 11: el nombre de la acción;
- línea 12: el número de valores que se emitirán;
- línea 13: una instancia de tipo [Func1<Integer, T>] que, a partir de un entero, crea un tipo T que será emitido por el observable (líneas 35 y 37);
- líneas 16-20: se pasan al constructor el nombre de la acción, el número de valores que se emitirán y la función de emisión;
- líneas 23-42: el código del proceso;
- línea 23: el método [call] recibe como parámetro al suscriptor del observable asociado al proceso;
- línea 28: el proceso emite sus elementos tras una espera de duración aleatoria;
- línea 32: emisión de un error;
- línea 37: una emisión normal;
- línea 41: emisión de la notificación de fin de emisión;
- líneas 25-38: la acción emite valores reales de nbValues tras un tiempo de espera aleatorio (línea 30);
- línea 35: el valor que se emitirá lo proporciona la función [func1] pasada como parámetro al constructor (línea 16);
Reestructuramos la clase [Process] (véase el párrafo 7.3.1) para que también pueda construirse con una acción con nombre. Le agregamos el siguiente constructor:
public Process(IProcessAction<T> na, Scheduler schedulerObserved, Scheduler schedulerObserver) {
// nombre del proceso=nombre de la acción
name = na.getName();
// acción --> observable
observable = Observable.create(na);
// hilo de ejecución del proceso observado
if (schedulerObserved != null) {
observable = observable.subscribeOn(schedulerObserved);
}
// hilo de observación del observador
if (schedulerObserver != null) {
observable = observable.observeOn(schedulerObserver);
}
}
- En la línea 1, el constructor admite 3 parámetros:
- la acción con nombre que se utilizará para construir el observable (línea 5);
- el programador del proceso observado (puede ser null);
- el programador del observador (puede ser null);
- línea 5: el observable se crea a partir de la acción pasada como parámetro;
El siguiente código [Exemple13] observa diferentes observables:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple13 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 1, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// proceso 3
Process<Integer> process3 = new Process<>(new ProcessAction01<Integer>("process3", 3, i -> i * 2), null,
Schedulers.computation());
// proceso 4
Process<Boolean> process4 = new Process<>(new ProcessAction01<Boolean>("process4", 4, i -> i % 2 == 0), null, null);
// suscripciones
ProcessUtils.subscribe(1, process1);
ProcessUtils.subscribe(1, process2);
ProcessUtils.subscribe(1, process3);
ProcessUtils.subscribe(1, process4);
}
}
- líneas 13-15: el proceso process1 genera un número real en un hilo de cálculo que será observado en otro hilo de cálculo;
- líneas 17-18: el proceso process2 genera 2 cadenas de caracteres en un hilo de cálculo y no se proporciona ninguna indicación sobre el hilo del observador. Los resultados muestran que la observación se realiza, por defecto, en el mismo hilo en el que se ejecuta el proceso;
- líneas 20-21: el proceso process3 genera 3 números enteros en un hilo no especificado que serán observados en un hilo de cálculo. Los resultados muestran que la ejecución del proceso se realiza, por defecto, en el hilo principal;
- línea 23: el proceso process4 genera 4 valores booleanos en un hilo no especificado, que serán observados en un hilo no especificado. Los resultados muestran que tanto la ejecución del proceso como su observación se realizan, por defecto, en el hilo principal;
El resultado de la ejecución de este código es el siguiente:
- el proceso process1 genera 1 número real (línea 4) en el hilo de cálculo [RxComputationThreadPool-4], el cual se observa en el hilo de cálculo [RxComputationThreadPool-3] (línea 6);
- el proceso process2 genera 2 cadenas de caracteres (líneas 12, 14) en el hilo de cálculo [RxComputationThreadPool-5], las cuales se observan en ese mismo hilo (líneas 13, 15);
- el proceso process3 genera 3 números enteros (líneas 21, 23, 25) en el hilo principal, los cuales se observan en el hilo de cálculo [RxComputationThreadPool-6] (líneas 22, 24, 28);
- el proceso process4 genera 4 valores booleanos (líneas 34, 36, 38, 40) en el hilo principal, los cuales se observan en ese mismo hilo principal (líneas 33, 35, 37, 39);
Se invita al lector a seguir lo anterior:
- el ciclo de vida del proceso observado y su hilo;
- el ciclo de vida de su observador y su hilo;
Gran parte del interés de las bibliotecas Rx radica en este multihilo que el desarrollador no tiene que gestionar por sí mismo.
7.5. Combinaciones de varios observables
7.5.1. Ejemplo 14: fusionar dos observables con [Observable.merge]
A continuación, presentamos los métodos estáticos de la clase [Observable] que permiten combinar varios observables en un observable de resultado.
El primer ejemplo de este tipo será el siguiente:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.ProcessAction01;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple14 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// fusión
Process<?> process12 = new Process<>("process12",
Observable.merge(process1.getObservable(), process2.getObservable()));
// suscripciones
ProcessUtils.subscribe(1, process12);
}
}
- líneas 15-17: un proceso llamado [process1] emitirá 3 números reales en un hilo de cálculo. También se observará en un hilo de cálculo;
- líneas 19-20: un proceso llamado [process2] emitirá 2 cadenas de caracteres en un hilo de cálculo. El hilo de observación no está predeterminado. Anteriormente vimos que, en este caso, el hilo de observación es el hilo de cálculo;
- línea 23: los dos procesos se fusionan, es decir, se crea un observable cuyos elementos provienen simultáneamente de ambos procesos. Para ello se utiliza el método estático [Observable.merge]:
 |
A diferencia de lo que podría sugerir el esquema anterior, durante la fusión, los elementos de un flujo 1 pueden intercalarse entre los elementos de un flujo 2. Esto es lo que muestran los resultados de la ejecución:
- línea 3: el proceso [process1] se ejecuta en el hilo de cálculo [RxComputationThreadPool-4];
- línea 4: el proceso [process2] se ejecuta en el hilo de cálculo [RxComputationThreadPool-5];
- línea 9: el proceso [process12] se observa en el hilo de cálculo [RxComputationThreadPool-3]. No conozco la regla que llevó a esta elección;
- líneas 9-11: se observa que el observador observa elementos de los dos procesos [process1] (línea 5) y [process2] (líneas 6, 7), aunque ninguno de los dos ha finalizado (hay una mezcla);
- El proceso [process12] finaliza (línea 17) cuando los dos procesos process1 y process2 han finalizado;
7.5.2. Ejemplo 15: concatenar dos observables con [Observable.concat]
Ahora analizaremos el siguiente código:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.ProcessAction01;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple15 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, Schedulers.computation());
// concat
Process<?> process12 = new Process<>("process12",
Observable.concat(process1.getObservable(), process2.getObservable()));
// suscripciones
ProcessUtils.subscribe(1, process12);
}
}
- líneas 15-17: un proceso llamado [process1] emitirá 3 números reales en un hilo de cálculo. También se observará en un hilo de cálculo;
- líneas 19-20: un proceso llamado [process2] emitirá 2 cadenas de caracteres en un hilo no especificado, en este caso el hilo principal por defecto. Se observará en un hilo de cálculo;
- línea 23: los dos procesos se concatenan, es decir, se crea un observable cuyos elementos provienen de ambos procesos. No hay mezcla de los valores emitidos. El proceso [process12] emitirá primero todos los valores del proceso [process1] y luego los del proceso [process2]. Para ello se utiliza el método estático [Observable.concat]:
 |
Los resultados de la ejecución son los siguientes:
- líneas 3-10: se ejecuta el proceso [process1] y el proceso [process12] emite los valores generados por [process1];
- línea 9: el proceso [process1] ha finalizado;
- líneas 11-17: el proceso [process2] se ejecuta y el proceso [process12] transmite los valores emitidos por [process2];
Hay una anomalía con el proceso process2: no se le había asignado un hilo de ejecución. Por lo tanto, cabría esperar que, por defecto, este fuera el hilo principal. Sin embargo, no es así. El hilo de ejecución fue el hilo de cálculo [RxComputationThreadPool-3] (línea 11). Por lo tanto, cuando no se especifica un hilo de ejecución o de observación, no se puede hacer ninguna suposición sobre qué hilo se elegirá.
7.5.3. Ejemplo 16: combinar dos observables con [Observable.zip]
Ahora analizaremos el siguiente código:
package dvp.rxjava.observables.exemples;
import java.util.Arrays;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.functions.FuncN;
import rx.schedulers.Schedulers;
public class Exemple16 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, null);
// función de combinación de los 2 procesos
FuncN<String> funcn = new FuncN<String>() {
@Override
public String call(Object... args) {
if (args.length == 2) {
return String.format("double=%s, string=%s", args[0], args[1]);
} else {
throw new RuntimeException("la fonction attend 2 paramètres exactement");
}
}
};
// compresión de los 2 procesos
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
// suscripciones
ProcessUtils.subscribe(1, process12);
}
}
- líneas 16-18: un proceso llamado [process1] emitirá 3 números reales en un hilo de cálculo. También será observado en un hilo de cálculo;
- líneas 20-21: un proceso llamado [process2] emitirá 2 cadenas de caracteres en un hilo no impuesto. El hilo de observación tampoco está impuesto;
- líneas 23-32: instanciación de un tipo [FuncN<String>] con una clase anónima. FuncN es una interfaz funcional:
 |
El método [FuncN.call] espera un array de objetos y devuelve un tipo R. La función [funcn] se utilizará para combinar los procesos process1 y process2 en ese orden. En el método [FuncN.call]:
- args[0] será un Double;
- args[1] será un String;
Aquí, el resultado de [funcn.call] será la cadena de caracteres de la línea 27. Para obtener este resultado no es necesario conocer los tipos de los argumentos del método call.
Ambos procesos se combinan de la siguiente manera:
// archivo zip de los 2 procesos
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
El método [Observable.zip] funciona de la siguiente manera:
 |
Se observa que:
- el primer argumento de zip es un Iterable<Observable>. En nuestro ejemplo, tenemos un parámetro efectivo de tipo List<Observable> formado por nuestros dos observables;
- el segundo argumento de zip es de tipo FuncN. En nuestro ejemplo, el parámetro efectivo es [funcn];
La ejecución arroja los siguientes resultados:
- líneas 7, 11: el proceso process12 emite dos elementos;
- línea 8: el elemento adicional emitido por el proceso process1, que no tiene contraparte en el proceso process2, no es emitido por el proceso de resultado process12;
Se observa que el proceso process2, al que no se le había asignado ni un hilo de ejecución ni un hilo de observación, utilizó el hilo principal para ambos.
7.5.4. Ejemplo 17: combinar dos observables con [Observable.combineLatest]
Ahora analizaremos el siguiente código:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple17 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null,
Schedulers.computation());
// combinación de los 2 procesos
Process<Double> process12 = new Process<>("process12",
Observable.combineLatest(process1.getObservable(), process2.getObservable(), (d1, d2) -> d1 + d2));
// suscripciones
ProcessUtils.subscribe(1, process12);
}
}
- líneas 14-16: un proceso llamado [process1] emitirá 3 números reales en un hilo de cálculo. También será observado en un hilo de cálculo;
- líneas 18-20: un proceso llamado [process2] emitirá 2 números reales en un hilo no asignado. Estos serán observados en un hilo de cálculo;
- línea 23: los dos observables se combinan con el siguiente método estático [Observable.combineLatest]:
 |
El observable [combineLatest] funciona de la siguiente manera: cuando uno de los dos observables emite un elemento E1, este elemento es combinado por [combineFunction] con el último elemento emitido por el otro observable.
La ejecución de este código da el siguiente resultado:
- línea 5: la emisión de process2 (56) se combina con el último elemento emitido por process1 (54, línea 4) y produce el resultado de la línea 7;
- línea 6: la emisión de process1 (51,6) se combina con el último elemento emitido por process2 (56, línea 5) y produce el resultado de la línea 8;
- línea 9: la emisión de process2 (261,8) se combina con el último elemento emitido por process1 (51,6, línea 6) y produce el resultado de la línea 12;
- línea 13: la emisión de process1 (80,39) se combina con el último elemento emitido por process2 (261,8, línea 9) y produce el resultado de la línea 15;
Aquí nos encontramos ante una variante del observable [zip], en la que, en esta ocasión, los elementos combinados no son necesariamente los elementos que ocupan la misma posición en los flujos. Cabe señalar aquí que el proceso process2, al que no se le había asignado un hilo de ejecución, se ejecutó en este caso en el hilo principal (línea 2).
7.5.5. Ejemplo 18: combinar dos observables con [Observable.amb]
Ahora analizaremos el siguiente código:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple18 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null, null);
// combinación de los 2 procesos
Process<Double> process12 = new Process<>("process12",
Observable.amb(process1.getObservable(), process2.getObservable()));
// suscripciones
ProcessUtils.subscribe(1, process12);
}
}
- líneas 14-16: un proceso llamado [process1] emitirá 3 números reales en un hilo de cálculo. También se observará en un hilo de cálculo;
- líneas 18-20: un proceso llamado [process2] emitirá 2 números reales en un hilo no restringido. Se observarán en un hilo no restringido;
- línea 22: los dos observables se combinan con el siguiente método estático [Observable.amb]:
 |
Como se muestra en el diagrama anterior, el observable [Observable.amb(Observable o1, Observable o2)] emite los elementos del observable que emite primero. Esto lo confirman los resultados del ejemplo presentado:
- línea 4: es el proceso process2 el que emite primero;
- líneas 8 y 12: el proceso process12 emite todos los elementos emitidos por el proceso process2 (líneas 4 y 11);
7.6. Cadena de procesamiento de un observable
7.6.1. Ejemplo 19: transformar un observable con [Observable.map]
En los ejemplos anteriores, hemos analizado diversas combinaciones de dos observables para obtener un tercer observable. Ahora presentaremos los métodos estáticos de la clase [Observable] que permiten realizar operaciones de transformación, filtrado y agregación sobre un observable. Aquí encontraremos métodos análogos a los de la clase [Stream] estudiados en el párrafo 5.
Nuestro primer ejemplo será el siguiente:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple19 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<String> process2 = new Process<>("process2",
process1.getObservable().map(d -> String.format("valeur-%s", d)));
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
- líneas 14-16: un proceso llamado process1 emitirá 3 números reales en un hilo de cálculo. También se observará en un hilo de cálculo;
- líneas 17-18: los números generados por process1 se convertirán en cadenas de caracteres en un proceso process2;
- línea 20: se observa process2;
El método [Observable.map] de la línea 18 es análogo al método [Stream.map] analizado en el párrafo 5.5:
 |
Los resultados del ejemplo son los siguientes:
- líneas 4, 5 y 8: las emisiones de process1. Son números reales;
- líneas 6, 7 y 10: las emisiones de process2 observadas. Son cadenas de caracteres;
7.6.2. Ejemplo-20: filtrar una variable observable con [Observable.filter]
El ejemplo será el siguiente:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple20 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().filter(i -> i % 2 == 0));
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
- líneas 11-12: un proceso llamado process1 emitirá los números enteros del 0 al 2 en un hilo de cálculo. También se observará en un hilo de cálculo;
- línea 14: los números generados por process1 se filtrarán para conservar en process2 únicamente los números pares;
- línea 20: se observa process2;
El método [Observable.filter] de la línea 18 es análogo al método [Stream.filter] analizado en el párrafo 5.4:
 |
Los resultados del ejemplo son los siguientes:
- líneas 4, 5 y 7: las transmisiones de process1;
- líneas 6 y 9: las emisiones de process2 observadas. Son los elementos de process1 los que son pares;
7.6.3. Ejemplo 21: transformar un observable con [Observable.flatMap]
El ejemplo será el siguiente:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple21 {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMap(i -> {
int value = i * 10;
return Observable.just(value, value + 1, value + 2);
}));
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
- líneas 12-13: un proceso llamado process1 emitirá los números enteros del 0 al 2 en un hilo de cálculo. También se observará en un hilo de cálculo;
- líneas 15-18: cada número n emitido por process1 se transforma en un observable que emite los 3 números (10*n, 10*n+1, 10*n+2). Si en la línea 15 se utilizara el método [map], process2 emitiría un tipo Observable<Integer> y no un tipo Integer. El método [flatMap] utilizado permite convertir (flatten) esta secuencia de elementos de tipo Observable<Integer> en una secuencia de elementos de tipo Integer compuesta por cada uno de los elementos de cada uno de los Observable<Integer>;
- línea 20: se observa process2;
El método [Observable.flatMap] de la línea 15 es análogo al método [Stream.flatMap] analizado en el párrafo 5.6.12:
 |
Los resultados del ejemplo son los siguientes:
- líneas 5-7: las tres transmisiones de process2 tras la transmisión de la línea 4 de process1;
- líneas 9-11: las tres emisiones de process2 tras la emisión de la línea 8 de process1;
- líneas 14-16: las tres emisiones de process2 tras la emisión de la línea 12 de process1;
El siguiente código muestra cómo crear un tipo Observable<Integer[]> a partir de process1 y [Exemple21b]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21b {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer[]> process2 = new Process<>("process2", process1.getObservable().map(i -> {
int value = i * 10;
return new Integer[] { value, value + 1, value + 2 };
}));
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
- línea 14: se utiliza el método [Observable.map];
- línea 16: que devuelve un tipo Integer[];
Los resultados son los siguientes:
- líneas 6, 7, 10: se ven los resultados de map;
Todas estas transformaciones de observables se pueden encadenar, ya que cada transformación produce un nuevo observable. Esto es lo que muestra el siguiente ejemplo [Exemple21c]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.schedulers.Schedulers;
public class Exemple21c {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMap(i -> {
int value = i * 10;
return Observable.just(value, value + 1, value + 2);
}).filter(i -> i % 2 == 0));
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
- líneas 15-18: el flatMap va seguido de un filter;
Los resultados de la ejecución son los siguientes:
- líneas 8-13: process2 solo generó los elementos pares procedentes de flatMap;
Un método similar a [flatMap] es el método [flatMapIterable], ilustrado por el siguiente ejemplo de [Exemple21d]:
package dvp.rxjava.observables.exemples;
import java.util.Arrays;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21d {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMapIterable(i -> {
int value = i * 10;
return Arrays.asList(value, value + 1, value + 2);
}).filter(i -> i % 2 == 0));
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
En la línea 16, en lugar de utilizar el método [flatMap], se utiliza el método [flatMapIterable]. En este caso, la función de transformación debe generar un tipo Iterable<T> (línea 18) en lugar de un tipo Observable<T>.
Se obtienen los mismos resultados que antes.
Volvamos a la definición del método [flatMap]:
 |
Como se ve arriba, se ha insertado un elemento azul [3] entre los dos elementos verdes [1-2]. Esto significa que, en su operación de aplanamiento de los Observable<T>, el método [flatMap] respeta el orden de emisión de estos diferentes observables internos. Esto se ilustra en el siguiente ejemplo [Exemple21e]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21e {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// proceso 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().flatMap(i -> process2.getObservable()));
// suscripciones
ProcessUtils.subscribe(1, process3);
}
}
- líneas 11-12: el proceso process1 genera los números enteros [0,1];
- líneas 14-15: el proceso process2 emite los números enteros [10,11,12];
- líneas 17-18: a cada elemento emitido por process1 se le asocia el observable del proceso process2. Esto significa que:
- al elemento [0] de process1 se le asociará un observable que emite los [10,11,12];
- lo mismo ocurre con el elemento 1;
Al final, se emitirán los 6 números [10, 11, 12, 10, 11, 12]. Queremos ver en qué orden.
Los resultados de la ejecución son los siguientes:
Se observa que el orden de emisión del proceso process3 fue: [10, 10, 11, 12, 11, 12] (líneas 11, 12, 14, 17, 19, 22). Por lo tanto, sí hubo una mezcla de los elementos emitidos por el proceso process2. Esto se puede evitar utilizando el método [concatMap] en lugar del método [flatMap]. Así lo muestra el siguiente código [Exemple21ef]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21ef {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// proceso 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().concatMap(i -> process2.getObservable()));
// suscripciones
ProcessUtils.subscribe(1, process3);
}
}
En la línea 18, se ha sustituido [flatMap] por [concatMap]. Los resultados de la ejecución son los siguientes:
Se observa que el orden de emisión del proceso process3 fue: [10, 11, 12, 10, 11, 12] (líneas 12-14, 17, 19, 22). Los elementos emitidos por el proceso process2 no se mezclaron.
Otra variante del método [map] es el método [switchMap]:
 |
En el ejemplo anterior, del observable [1] surgen otros tres observables [2] de dos elementos, los cuales luego se aplanan como en [flatMap] y [3]. Se puede observar que el resultado tiene 5 elementos y no 6. Esto se debe a que, antes de que el segundo observable emita su elemento n.º 2, [6], el tercer observable emite su primer elemento, [5], lo que hace que el segundo observable sea descartado. Por lo tanto, no se encuentra el elemento [6] en el observable resultante [3].
Para ilustrar [switchMap], utilizaremos el siguiente ejemplo [Exemple21eg]:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple21eg {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// proceso 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().switchMap(i -> process2.getObservable()));
// suscripciones
ProcessUtils.subscribe(1, process3);
}
}
La ejecución del ejemplo arroja los siguientes resultados:
- process1 emite 2 elementos que dan lugar a 2 observables process2 de 3 elementos;
- línea 14: el observador recibe el elemento n.º 0 emitido por el primer observable process2 de la línea 6;
- línea 15: el observador recibe el elemento n.º 0 emitido por el segundo observable process2 en la línea 13. No se explica por qué no recibió antes los elementos 1 y 2 emitidos por el primer observable process2 en las líneas 7 y 8. Sea como fuere, el primer observable process2 queda descartado;
- al final, el observador solo ve 4 elementos (líneas 14, 15, 17, 20) en lugar de los 6 que se emitieron;
7.6.4. Ejemplos-22: otros métodos de la clase [Observable]
La clase [Observable] retoma numerosos métodos de la clase [Stream] con un funcionamiento similar. A continuación se presentan algunos de ellos. Nos limitaremos a proporcionar el código y sus resultados.
[Exemple22a - take=limit]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22a {
public static void main(String[] args) throws InterruptedException {
// proceso
Process<Integer> process = new Process<>("process", Observable.range(1, 10).take(3));
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
Resultados
[Exemple22b - takeLast]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22b {
public static void main(String[] args) throws InterruptedException {
// proceso
Process<Integer> process = new Process<>("process", Observable.range(1, 10).takeLast(2));
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
resultados
[Exemple22c - skip]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22c {
public static void main(String[] args) throws InterruptedException {
// procesos
Process<Integer> process = new Process<>("process", Observable.range(1, 10).skip(5).take(2));
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
resultados
[Exemple22d - reduce]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22d {
public static void main(String[] args) throws InterruptedException {
// procesos
Process<Integer> process = new Process<>("process", Observable.range(1, 10).reduce(0, (i, a) -> i + a));
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
- línea 10: calcula la suma de los elementos del observable. El resultado es un observable que emite dicha suma;
resultados
[Exemple22e - all]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22e {
public static void main(String[] args) throws InterruptedException {
// procesos
Process<Boolean> process = new Process<>("process", Observable.range(1, 10).all(i -> i > 10));
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
- línea 10: devuelve un Observable<Boolean> que emite el elemento true, si el predicado del método [all] es verdadero para todos los elementos; de lo contrario, false;
resultados
[Exemple22f - count]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22f {
public static void main(String[] args) throws InterruptedException {
// procesos
Process<Integer> process = new Process<>("process", Observable.range(1, 10).count());
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
- línea 10: [Observable.count] crea un observable de 1 elemento que es la suma de los elementos observados;
resultados
[Exemple22g - distinct]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
public class Exemple22g {
public static void main(String[] args) throws InterruptedException {
// procesos
Process<Integer> process = new Process<>("process", Observable.just(1, 2, 1, 3).distinct());
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
resultados
[Exemple22h - groupBy, asObservable]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.observables.GroupedObservable;
public class Exemple22h {
public static void main(String[] args) throws InterruptedException {
// procesos
Observable<GroupedObservable<Boolean, Integer>> obs = Observable.range(1, 10).groupBy(i -> i % 2 == 0);
Process<Integer> process = new Process<>("process", obs.concatMap(g -> g.asObservable()));
// suscripciones
ProcessUtils.subscribe(1, process);
}
}
- línea 11: el método [groupBy] agrupa los 10 elementos emitidos en 2 grupos, los números pares y los números impares. El resultado es un tipo Observable<GroupedObservable<Boolean, Integer>>, es decir, un observable cuyos elementos son de tipo GroupedObservable<Boolean, Integer>, donde Boolean es el tipo de la clave del grupo (false, true en este caso) y que también es el tipo del resultado de la lambda pasada como parámetro al método [groupBy], y Integer el tipo de los elementos del grupo;
- línea 12: el tipo GroupedObservable cuenta con un método [asObservable] que permite crear un observable a partir de este tipo. Por lo tanto, tendremos dos tipos Observable<Integer>, uno para los números pares y otro para los impares. A partir de estos dos observables, el método [concatMap] creará uno solo;
resultados
[Exemple22i - timestamp]
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
import rx.schedulers.Timestamped;
public class Exemple22i {
public static void main(String[] args) throws InterruptedException {
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// proceso 2
Process<Timestamped<Integer>> process2 = new Process<>("process2", process1.getObservable().timestamp());
// suscripciones
ProcessUtils.subscribe(1, process2);
}
}
- En la línea 15, el método [timestamp] asocia una hora a cada elemento de la variable observable procesada;
resultados
En este ejemplo, es difícil determinar qué representa la información timestamp:
- líneas 4-5: se observa que el elemento 1 de process1 se emitió 139 ms después del elemento 0;
- líneas 6 y 7: se observa que el elemento 1 de process2 fue detectado 234 ms después del elemento 0;
- líneas 5 y 8: se observa que el elemento 2 de process1 se emitió 33 ms después del elemento 1;
- líneas 7 y 10: se observa que el elemento 2 de process2 fue observado 37 ms después del elemento 1;
Estos desfases se deben a que los subprocesos de observación y de ejecución de los observables no son los mismos. Si reemplazamos las líneas 12-13 por las siguientes (Ejemplo22j):
// proceso 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
null);
- líneas 2-3: no se impone el hilo de observación. Sabemos que, en este caso, el observable se observa donde se ejecuta;
Esto da los siguientes resultados:
- líneas 4 y 6: el proceso process1 emite su elemento n.º 1 587 ms después de su elemento n.º 0;
- líneas 5 y 7: el observador observa estos dos elementos con un intervalo de 586 ms;
- líneas 6 y 8: el proceso process1 emite su elemento n.º 2 396 ms después de su elemento n.º 1;
- líneas 7 y 9: el observador observa estos dos elementos con un intervalo de 396 ms;
En este caso, los valores de timestamp son coherentes: representan correctamente la fecha de emisión del elemento.
7.7. Los programadores
7.7.1. Ejemplo 23: el programador [Schedulers.computation]
Ahora analizaremos los programadores de ejecución. La observación se realizará en el hilo de ejecución.
El tema de los programadores es un poco confuso. Los diferentes programadores se presentan en esta pregunta en el sitio web de StackOverflow [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
Intentaremos ilustrar el uso de estos diferentes programadores con ejemplos. El primero ilustra el programador [Schedulers.computation]:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple23 {
public static void main(String[] args) throws InterruptedException {
// procesos
@SuppressWarnings("unchecked")
Process<Double> processes[] = new Process[10];
for (int i = 0; i < processes.length; i++) {
processes[i] = new Process<>(
new ProcessAction01<Double>(String.format("process%s", i), 1, value -> new Random().nextInt(100) * 1.2),
Schedulers.computation(), null);
}
// suscripciones
ProcessUtils.subscribe(1, processes);
}
}
- líneas 14-19: se crea una matriz de 10 procesos que se ejecutan en un hilo de cálculo;
- línea 17: cada proceso genera un número real aleatorio;
- línea 21: se realiza una suscripción a todos estos procesos;
Los resultados son los siguientes:
- líneas 2-10: los primeros 8 procesos se inician en 8 hilos diferentes (la máquina utilizada tiene 8 núcleos). Se puede observar que todos comienzan aproximadamente al mismo tiempo;
- líneas 17-19: 3 procesos finalizan y, por lo tanto, liberan 3 subprocesos;
- líneas 23-24: los dos últimos procesos pueden entonces iniciarse utilizando 2 de los subprocesos así liberados;
Por lo tanto, cabe destacar que el programador [Schedulers.computation] proporciona un conjunto de n hilos, donde n es el número de núcleos de la máquina. Los hilos se ejecutan en paralelo en estos núcleos.
7.7.2. Ejemplo 24: el programador [Schedulers.io]
Ejecutamos el código anterior con el programador [Schedulers.io]:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple24 {
public static void main(String[] args) throws InterruptedException {
// procesos
@SuppressWarnings("unchecked")
Process<Double> processes[] = new Process[10];
for (int i = 0; i < processes.length; i++) {
processes[i] = new Process<>(
new ProcessAction01<Double>(String.format("process%s", i), 1, value -> new Random().nextInt(100) * 1.2),
Schedulers.io(), null);
}
// suscripciones
ProcessUtils.subscribe(1, processes);
}
}
- línea 18: los procesos se ejecutan con los hilos del programador [Schedulers.io];
Esto da los siguientes resultados:
- líneas 2-10: los 10 procesos se inician cada uno en un hilo diferente. A diferencia del caso anterior, se pudieron iniciar todos los procesos. Se observa que estos inicios tardan 6 ms, mientras que anteriormente habían tardado 1 ms;
- líneas 13-18: los observables emiten uno tras otro y no de manera casi paralela, como había sido el caso anteriormente;
¿Cuál es la diferencia entre los programadores [Schedulers.io] y [Schedulers.computation]? Se puede encontrar una respuesta en URL [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
7.7.3. Ejemplo 25: el programador [Schedulers.newThread]
Ejecutamos el código anterior con el programador [Schedulers.newThread]:
package dvp.rxjava.observables.exemples;
import java.util.Random;
import dvp.rxjava.observables.utils.Process;
import dvp.rxjava.observables.utils.ProcessAction01;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.schedulers.Schedulers;
public class Exemple25 {
public static void main(String[] args) throws InterruptedException {
// procesos
@SuppressWarnings("unchecked")
Process<Double> processes[] = new Process[10];
for (int i = 0; i < processes.length; i++) {
processes[i] = new Process<>(
new ProcessAction01<Double>(String.format("process%s", i), 1, value -> new Random().nextInt(100) * 1.2),
Schedulers.newThread(), null);
}
// suscripciones
ProcessUtils.subscribe(1, processes);
}
}
Los resultados obtenidos son los mismos que con el programador [Schedulers.io]:
En URL y [http://stackoverflow.com/questions/33415881/retrofit-with-rxjava-schedulers-newthread-vs-schedulers-io], se explica que el programador [Schedulers.io] proporciona un grupo de subprocesos, algo que no hace el programador [Schedulers.newThread]. Un grupo de subprocesos creará automáticamente un número n de subprocesos. Los asignará a los procesos que los necesiten. Cuando estos procesos finalizan, sus subprocesos no se eliminan, sino que regresan al grupo y pueden ser reutilizados por otro proceso. Esto es más eficiente que crear y eliminar subprocesos constantemente. Por lo tanto, se puede considerar que es preferible utilizar el programador [Schedulers.io].
7.7.4. Ejemplo 26: los programadores [Schedulers.immediate, Schedulers.trampoline]
Volvamos a la explicación dada para estos dos programadores:
|
La explicación es bastante sencilla de entender, pero cuando uno quiere ilustrarla, se da cuenta de que no la ha entendido. Fue el libro [Learning Reactive Programming With Java 8] el que me permitió crear un ejemplo que retoma uno encontrado en ese libro, pero lo simplifica. Es el siguiente:
package dvp.rxjava.observables.exemples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.function.Consumer;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Scheduler;
import rx.Scheduler.Worker;
import rx.functions.Action0;
import rx.schedulers.Schedulers;
public class Exemple26 {
public static void main(String[] args) throws InterruptedException {
// un programador
Scheduler scheduler = Schedulers.immediate();
// un trabajador de este programador
Worker worker = scheduler.createWorker();
// un tipo Action0 para ejecutar en el trabajador
Action0 action02 = new Action0() {
@Override
public void call() {
// registro de acción02
ProcessUtils.showInfos.accept("action02");
}
};
// un tipo Action0 para ejecutar en el trabajador
Action0 action01 = new Action0() {
@Override
public void call() {
// se programa una nueva acción en el mismo trabajador
worker.schedule(action02);
// registro de la acción 01
ProcessUtils.showInfos.accept("action01");
}
};
// la acción 01 está programada en el trabajador
worker.schedule(action01);
}
// visualizaciones
static Consumer<String> showInfos = message -> System.out.printf("%s ------Thread[%s] ---- Time[%s]%n", message,
Thread.currentThread().getName(), new SimpleDateFormat("ss:SSS").format(new Date()));
}
- línea 17: un programador. Será [Schedulers.immediate], como aquí, o bien [Schedulers.trampoline] más adelante;
- línea 19: se pueden ejecutar acciones del tipo Action0 (líneas 21, 20) en los trabajadores del programador. El método [Scheduler.createWorker] permite crear un trabajador. El método [Worker.schedule(Action0)] permite que un trabajador ejecute un tipo Action0;
- líneas 21-27: una primera acción llamada [action02] que será ejecutada (línea 40) por el trabajador de la línea 19;
- líneas 30-38: una segunda acción llamada [action01]. Tiene la particularidad de hacer que se ejecute la acción action02 en el mismo trabajador que ella (línea 34). Ahí radica la diferencia entre [Schedulers.immediate] y [Schedulers.trampoline]:
- si el programador es [Schedulers.immediate], entonces, en la línea 34, la acción action02 se ejecutará de inmediato (de ahí el nombre del programador) y la acción action01 que se está ejecutando se interrumpirá. Entonces aparecerá el mensaje de la línea 25. Una vez finalizada la acción action02, se reanudará la acción action01 y aparecerá el mensaje de la línea 36;
- si el programador es [Schedulers.trampoline], entonces, en la línea 34, la acción action02 se pone en espera. No se ejecutará hasta que la tarea en curso action01 haya finalizado. Entonces aparecerá el mensaje de la línea 36. Una vez finalizada la acción action01, se ejecutará la acción action02 y veremos el mensaje de la línea 25;
La ejecución del código anterior arroja los siguientes resultados:
Si en la línea 17 se utiliza el programador [Schedulers.trampoline], se obtienen los resultados opuestos:
Dicho esto, es difícil establecer una relación con los observables. No he encontrado ningún ejemplo convincente que pudiera demostrar la ventaja de ejecutar un observable en uno de estos dos hilos. Sin embargo, aquí hay uno, aunque no me parece nada natural:
package dvp.rxjava.observables.exemples;
import dvp.rxjava.observables.utils.ProcessUtils;
import rx.Observable;
import rx.Scheduler.Worker;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
public class Exemple27 {
public static void main(String[] args) throws InterruptedException {
// Trabajador
Worker worker = Schedulers.immediate().createWorker();
// Worker worker = Schedulers.trampoline().createWorker();
// observable 1 en el worker
worker.schedule(() -> Observable.range(1, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
// observable 2 en el mismo worker
worker.schedule(() -> Observable.range(100, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
}
}));
}
}));
}
}
- líneas 13-14: se crea un trabajador a partir de uno de los dos programadores [Schedulers.immediate] y [Schedulers.trampoline];
- línea 16: se programa un primer observable obs1 en este trabajador para emitir los números [1,2]
- línea 22: cada vez que se observa un elemento de este observable obs1, se inicia la observación de un segundo observable obs2 en el mismo trabajador para generar los números [100,101];
Con el programador [Schedulers.immediate], se obtienen los siguientes resultados:
Mientras que con el programador [Schedulers.trampoline], se obtienen los siguientes resultados:
7.8. Conclusion
Aún queda mucho por hacer. Para profundizar en la biblioteca RxJava, se invita al lector a continuar su formación con las referencias proporcionadas al inicio de este documento. A pesar de todo, contamos con las bases para utilizar RxJava en los entornos Swing y Android. Eso es lo que vamos a mostrar ahora.