7. A biblioteca RxJava
A biblioteca RxJava baseia-se no seguinte conceito: um fluxo de elementos do tipo T Observable<T> é observado por um ou mais assinantes (assinantes, observadores, consumidores) Subscriber<T>. A biblioteca RxJava permite que o fluxo Observable<T> seja executado em um thread T1 e seu observador Subscriber<T> em um thread T2, sem que o desenvolvedorprecise se preocupar em gerenciar o ciclo de vida dessas threads e com problemas naturalmente complexos, como o compartilhamento de dados entre threads e a sincronização entre elas para executar uma tarefa global. Assim, ela facilita a programação assíncrona.
Um fluxo Observable<T> produz elementos do tipo T, observáveis à medida que são produzidos. Se o observador e o observável (termo que, por conveniência, designa o tipo Observable<T>) estiverem na mesma thread, então o observável só poderá produzir o elemento (i+1) quando o observador tiver consumido o elemento i. Há poucos casos em que essa arquitetura seja interessante. Se o observador e o observável não estiverem no mesmo thread, então o observável e seu observador têm comportamentos autônomos: o observável produz em seu próprio ritmo e o observador consome em seu próprio ritmo. É aí que reside o interesse da biblioteca. Até agora, sempre falamos de um único observador. Na realidade, um observável pode ter qualquer número de observadores.
A biblioteca RxJava é particularmente bem adaptada à arquitetura apresentada no parágrafo 2 da seção “Descoberta” e que relembramos aqui:
- em [1], uma camada de serviço fornece serviços, alguns dos quais demoram a ser obtidos (por exemplo, solicitações de rede);
- essa camada de serviços é invocada por uma interface gráfica [1] (Swing, Android, JavaFx). Se a camada de serviços for executada no mesmo thread que o método [swing] que a utiliza, a interface gráfica fica congelada (não responde) enquanto aguarda o resultado do serviço;
- em [2], uma fina camada de adaptação implementada com RxJava permite apresentar à camada gráfica uma implementação assíncrona do mesmo serviço: este pode ser executado em uma thread diferente daquela do método da camada gráfica que o invoca. Nesse caso, a interface gráfica [3] permanece responsiva: o usuário pode continuar a interagir com ela, por exemplo, iniciar uma nova solicitação de rede paralelamente à primeira e, acima de tudo, é possível oferecer a ele a possibilidade de cancelar processamentos muito demorados, o que seria impossível se a interface gráfica estivesse congelada;
- a chamada [4] é síncrona, enquanto a chamada [5-6] é assíncrona;
Nessa arquitetura, a camada [2] oferece serviços que retornam tipos Observable<T>, aos quais os métodos da camada gráfica [3] podem se inscrever. Um serviço da camada [2], então, fornece seus resultados um por um, e a camada [3] pode reagir a cada um deles, por exemplo, atualizando um ou mais componentes da interface gráfica.
A classe Observable<T> possui dezenas de métodos. Essa é uma das dificuldades da biblioteca: ela é muito rica e é difícil compreender todas as suas possibilidades. Apresentaremos algumas delas. O domínio dos demais métodos virá com o tempo.
7.1. Criando observáveis e assinando-os
7.1.1. Exemplo-01: o método [Observable.from]
 |
Consideremos o código a seguir:
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) {
// observáveis inteiros
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");
}
});
}
}
- linha 12: cria-se um tipo Observable<Integer> a partir de uma lista de inteiros.
A classe Observable<T> é um fluxo de elementos do tipo T que podem ser observados, de preferência de forma assíncrona, mas não necessariamente, à medida que são produzidos. Sua definição é a seguinte:
 |
Como já foi mencionado, a classe Observable<T> possui várias dezenas de métodos. Alguns são semelhantes aos da classe Stream<T>, estudada no parágrafo 5. A documentação da RxJava inclui “diagramas de mármore” [2] que ilustram o funcionamento desses métodos:
- a linha 3 ilustra as emissões do observável ao longo do tempo;
- o método [4] é aplicado aos elementos emitidos pelo observável. Ele geralmente produz um novo observável;
- a linha 5 mostra o novo observável obtido;
O método [Observable.from] tem a seguinte assinatura:
 |
O método estático [Observable.from] permite criar um Observable<T> a partir de uma coleção de elementos do tipo T. É uma maneira muito simples de começar a trabalhar com observáveis. A linha:
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
irá, portanto, emitir três elementos. Ela não os emite imediatamente. Ela os emitirá na íntegra sempre que um observador se registrar. Isso é o que chamamos de observável frio. O observável reemite seus elementos para cada novo assinante.
Pode-se considerar a instrução anterior como uma ação de configuração do observável. Este é configurado uma vez e executado n vezes se houver n assinantes.
Como se inscreve?
Uma maneira de fazer isso é usar o método [Observable.subscribe], cuja definição utilizada aqui é a seguinte:
 |
- o primeiro parâmetro [Action1<T> onNext] (ver parágrafo 6.2) do método é o método a ser executado quando o observável emite um novo elemento T;
- o segundo parâmetro [Action1<Throwable> onError] do método é o método a ser executado quando o observável lança uma exceção;
- o terceiro parâmetro [Action0 onComplete] (ver parágrafo 6.1) do método é o método a ser executado quando o observável lança uma exceção;
- o método retorna um tipo [Subscription];
O tipo [Subscription] representa uma assinatura do observável. Sua definição é a seguinte:
 |
O interesse dessa interface [1] reside em seu método [2], que permite cancelar uma assinatura.
No nosso exemplo, o código da assinatura do observável é o seguinte:
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");
}
});
- linha 1: o resultado do tipo [Subscription] é ignorado;
- linhas 1-15: os três parâmetros são instâncias de classes anônimas. Também utilizaremos lambdas. A vantagem das classes anônimas é que fica claro quais são os tipos de dados esperados pelo único método dessas classes;
- linhas 2-5: implementação do primeiro parâmetro do tipo [Action1<Integer>];
- linhas 6-10: implementação do segundo parâmetro do tipo [Action1<Throwable>];
- linhas 11-15: implementação do terceiro parâmetro do tipo [Action0];
O código completo é o seguinte:
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) {
// observáveis de inteiros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// assinatura
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");
}
});
}
}
O observável da linha 12 começa a emitir seus três elementos assim que o método [subscribe] é chamado na linha 14. A partir desse momento:
- a cada elemento emitido, as linhas 15 a 18 são executadas.
- ao final dos 3 elementos, as linhas 24 a 29 são executadas;
- as linhas 19 a 24 nunca serão executadas, pois o observável não lança nenhuma exceção neste caso;
Por padrão, o observável e o observador são executados na mesma thread. Existem alguns observáveis predefinidos que são executados em uma thread diferente da thread principal (neste caso, a thread do método main), mas, para a maioria deles, esse não é o caso. Portanto, aqui tudo ocorre na thread do método [main]:
- o observável emite o elemento 1;
- as linhas 15 a 18 são executadas e exibem esse elemento;
- o observável emite o elemento 2;
- as linhas 15 a 18 são executadas e exibem esse elemento;
- o observável emite o elemento 3;
- as linhas 15 a 18 são executadas e exibem esse elemento;
- o observável emite a notificação [completed];
- as linhas 24 a 29 são executadas;
É isso que mostram os resultados obtidos:
A classe [Exemple02] herda de [Exemple01], utilizando, desta vez, funções lambda como parâmetros do método [Observable.subscribe]:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
public class Exemple02 {
public static void main(String[] args) {
// observáveis de inteiros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// assinatura
obs1.subscribe(
(integer) -> System.out.printf("next : %s%n", integer),
(th) -> System.out.println(th),
() -> System.out.println("completed"));
}
}
7.1.2. Exemplo-03: a classe Observer
 |
O método [Observable.subscribe], que permite assinar um observável, possui diversas versões, entre as quais a seguinte:
package dvp.rxjava.observables;
import java.util.Arrays;
import rx.Observable;
import rx.Observer;
public class Exemple03 {
public static void main(String[] args) {
// observáveis de inteiros
Observable<Integer> obs1 = Observable.from(Arrays.asList(1, 2, 3));
// assinatura
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);
}
});
};
}
Na linha 13, em vez de passar três parâmetros para o método [subscribe], passa-se a ele um tipo [Observer], conforme segue:
 |
O tipo [Observer] é uma interface com três métodos:
- [onNext(T t)], que é chamado sempre que o observável emite um elemento t;
- [onError(Throwable th)], que é chamado quando o observável lança uma exceção th;
- [onCompleted], que é chamado quando o observável indica que concluiu a emissão;
O funcionamento do código é semelhante ao explicado anteriormente. Obtêm-se os seguintes resultados:
7.1.3. Exemplo-04: o método [Observable.create]
 |
O método estático Observable.create é definido da seguinte forma:
 |
- o método [create] retorna um tipo Observable<T>;
- o parâmetro do método [create] é uma função do tipo [Observable.OnSubscribe<T>] definida da seguinte forma:
 |
O tipo [Observable.OnSubscribe<T>] é uma interface funcional que, por sua vez, estende a interface funcional [Action1<Subscriber<? super T>>]. O método [call] dessa interface espera um tipo [Subscriber] (assinante, subscritor, observador) definido da seguinte forma:
 |
Vemos em [1] que a classe [Subscriber<T>] implementa a interface [Observer<T>] apresentada no parágrafo 7.1.2.
Por fim, o método [<T> Observable.create]:
- recebe como parâmetro uma instância do tipo [Observable.OnSubscribe<T>] com o único método de assinatura: void call(Subscriber<T> s). O tipo [Subscriber<T>] estende o tipo [Observer<T>] e, portanto, dispõe dos métodos onNext, onError, onCompleted;
- retorna um tipo Observable<T>;
O método [<T> Observable.create] retorna um observável configurado. Ainda não houve nenhuma emissão de elementos. Quando um assinante [Subscriber<T> s] se inscreve nesse observável, o método [void call(s)] da função passada como parâmetro do método [<T> Observable.create] é então chamado. Sua função é emitir elementos t do tipo T e chamar o método [s.onNext(t)] do observador a cada emissão. Quando este for concluído, o método [s.onCompleted(t)] do observador deve ser chamado e o método [call] deve ser encerrado. Se o método [call] encontrar uma exceção th, o método [s.onError(th)] do observador deve ser chamado e o método [call] deve ser encerrado;
Para ilustrar esse funcionamento complexo, utilizaremos o seguinte 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) {
// configuração observável de números reais
Observable<Double> obs1 = Observable.create(new Observable.OnSubscribe<Double>() {
@Override
public void call(Subscriber<? super Double> subscriber) {
for (int i = 0; i < 3; i++) {
// emissão do elemento i
subscriber.onNext(new Random((i + 1)).nextDouble());
}
// fim da transmissão
subscriber.onCompleted();
}
});
// assinatura e, portanto, emissão
obs1.subscribe((d) -> System.out.printf("onNext %s%n", d), (th) -> System.out.printf("onError %s%n", th),
() -> System.out.println("onCompleted"));
}
}
- linha 11: cria-se um observável que emite tipos Double;
- linhas 11-21: o parâmetro do método [create] é instanciado com uma classe anônima que possui o único método [call] das linhas 12-20. O observável criado na linha 11 está pronto para emitir, mas só emitirá quando um observador for registrado;
- linhas 13-21: o método [call] recebe a referência de um observador;
- linhas 14-17: emissão de 3 elementos para o observador;
- linha 19: notificação de fim de transmissão ao observador;
- linhas 23-24: assinatura do observável da linha 11. Implementamos os três parâmetros [onNext, onError, onCompleted] do método [subscribe] por meio de três funções lambda. Essa assinatura criará o assinante [Subscriber<Double>], que será passado para o método [call] da linha 13. A emissão de elementos terá então início;
- tudo ocorre no mesmo thread: observável e observador;
Obtêm-se os seguintes resultados:
O método [Observable.create] permite criar um observável a partir de qualquer fenômeno. Foi esse método que utilizamos no parágrafo 2 da seção “Descoberta”, para transformar uma interface síncrona em uma interface assíncrona.
7.1.4. Exemplo-05: refatoração de [Exemple-04]
 |
O exemplo a seguir apresenta uma nova versão do 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) {
// configuração de um observável de números reais
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++) {
// em espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// erro
subscriber.onError(e);
}
// ação
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// concluído
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// um assinante
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));
}
};
// assinatura
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()));
}
}
- linha 56: a nova versão do método estático [Observable.subscribe] aceita como parâmetro o tipo [Subscriber], que apresentamos no parágrafo anterior;
- linhas 37-52: o subscritor (assinante, observador). Ele implementa a interface Observer com seus três métodos onNext, onError e onCompleted;
- linhas 61-64: a partir de agora, vamos nos concentrar nos threads nos quais o observável e seu observador são executados;
- linha 62: o nome do thread;
- linha 63: a hora atual expressa em segundos e milissegundos. Isso nos permitirá acompanhar, ao longo do tempo, a emissão de elementos pelo observável e seu processamento pelo observador;
- esse código tem a mesma funcionalidade que o código anterior. Simplesmente refatoramos este último;
Os resultados obtidos são os seguintes:
- linha 1 dos resultados: antes da linha 56 do código, ainda não aconteceu nada. O observável foi simplesmente configurado;
- linha 2 dos resultados: a linha 56 do código provoca a chamada do método [call] da linha 15. Na linha 3, o número real 80,39 é enviado ao observador;
- linha 4: o observador recebe o número enviado;
- linhas 5-8: o processo anterior se repete duas vezes;
- linha 9: o observável envia a notificação de fim de transmissão;
- linha 10: o observador a recebe;
- linha 11: exibida pela linha 57 do código;
Vemos, portanto, que apenas a linha 56 de assinatura provocou a exibição das linhas 2 a 10 dos resultados. Ao começar a trabalhar com a biblioteca RxJava, questionamo-nos sobre como as coisas se encadeiam entre si e, em especial, sobre as ligações que unem o observador e o observável. Vemos aqui que a linha 56, a subscrição do observável,
- provocou a emissão de todos os elementos do observável;
- que o observável e o observador são executados no mesmo thread;
- que, por causa disso, observa-se a sequência: emissão do elemento i, observação do elemento i, emissão do elemento (i+1), observação do elemento (i+1), ...
Lembramos que o emissor aguardava antes de emitir seus elementos:
// em espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// erro
subscriber.onError(e);
}
onde i na linha 3 representa o número da emissão (0<=i<3). Se observarmos os horários de emissão dos elementos do observável:
- linhas 2, 3: o elemento 0 foi transmitido cerca de 500 ms após o início da assinatura;
- linhas 3, 5: o elemento 1 foi transmitido cerca de 400 ms após o elemento 0;
- linhas 5, 7: o elemento 2 foi emitido cerca de 300 ms após o elemento 1;
7.2. Thread de execução, thread de observação
7.2.1. Exemplo-06: observável e observador em um thread diferente de [main]
 |
Reestruturamos o exemplo anterior da seguinte maneira [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) {
// barreira de proteção
CountDownLatch latch = new CountDownLatch(1);
// configuração de um observável de valores reais
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++) {
// em espera
try {
Thread.sleep(500 - i * 100);
} catch (InterruptedException e) {
// erro
subscriber.onError(e);
}
// ação
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// concluído
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// um assinante
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// abaixando a barreira
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));
}
};
// continuação da configuração observável
obs1 = obs1.subscribeOn(Schedulers.computation());
// subscrição
showInfos("avant souscription");
obs1.subscribe(subscriber);
// espera em frente à barreira
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()));
}
}
- linha 16: criamos um semáforo com um objeto do tipo [CountDownLatch]. Esse objeto serve para sincronizar threads entre si. Aqui, ele é inicializado com o valor 1, que chamaremos de valor do semáforo. Uma thread entra em espera no semáforo por meio de uma operação:
latch.await();
A thread fica bloqueada se o valor do guard-barrier for >0. Uma thread pode aumentar ou diminuir o valor interno do guard-barrier. Na linha 48, o valor do guard-barrier é decrementado em 1.
- linha 63: o observável é configurado para ser executado em um thread fornecido pelo agendador [Schedulers.computation()]. Esse agendador pode fornecer tantos threads quantos forem os núcleos da máquina de execução. O parágrafo sobre o exemplo de aplicação mostrou o uso de outros agendadores (ver parágrafo 2.8);
O princípio do código é o seguinte:
- o método [main] é executado na thread principal (main);
- linha 66: inicia a emissão de elementos do observável. Estes serão emitidos em uma thread diferente da thread principal;
- linha 70: a thread principal fica bloqueada porque o bloqueador tem o valor 1 (ver linha 16). Ela só poderá continuar quando esse valor passar para 0. Isso ocorre na linha 48. É o observador que desativa o bloqueador ao receber a notificação de que o observável concluiu suas emissões;
A execução produz os seguintes resultados:
- linha 1: a assinatura será realizada;
- linha 2: isso aciona a execução do método [call] na thread [RxComputationThreadPool-1]. Agora temos uma execução paralela com duas threads;
- linha 3: por um motivo ainda não esclarecido, a thread [RxComputationThreadPool-1] cedeu a controle. A thread [main] assume então o controle e é bloqueada pelo guard-rail (linha 70 do código). A partir desse momento, apenas a thread [RxComputationThreadPool-1] pode operar;
- linhas 4-11: observa-se o comportamento observado anteriormente entre o observável e seu observador, mas agora tudo ocorre na thread [RxComputationThreadPool-1];
- linhas 12-13: o observador abaixou a barreira (linha 48 do código) e o thread [RxComputationThreadPool-1] foi encerrado. O thread [main] assume o controle e exibe duas mensagens;
7.2.2. Exemplo-07: observável e observador em dois threads diferentes
|
Modificamos o exemplo anterior da seguinte maneira:
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) {
// guarda da barreira
CountDownLatch latch = new CountDownLatch(1);
// configuração de um observável de números reais
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) {
// erro
subscriber.onError(e);
}
// ação
double value = new Random().nextInt(100) * 1.2;
showInfos(String.format("Observable.call onNext(%s)", value));
subscriber.onNext(value);
}
// concluído
showInfos(String.format("Observable.call onCompleted"));
subscriber.onCompleted();
}
});
// um assinante
Subscriber<Double> subscriber = new Subscriber<Double>() {
@Override
public void onCompleted() {
showInfos("Subscriber.onCompleted");
// abaixando a barreira
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));
}
};
// continuação da configuração observável
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
// subscrição
showInfos("avant souscription");
obs1.subscribe(subscriber);
// aguardando que a barreira suba
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()));
}
}
O código é idêntico ao do exemplo anterior, exceto pela linha 63:
obs1 = obs1.subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation());
que configura o observável (subscribeOn) e o observador (observeOn) para serem executados em uma das threads fornecidas pelo agendador [Schedulers.computation()].
Os resultados obtidos são os seguintes:
É possível observar os seguintes pontos:
- o observável é executado na thread [RxComputationThreadPool-4] (linhas 3-4, 6, 8-9);
- o observador é executado na thread [RxComputationThreadPool-3] (linhas 5, 7, 10-11);
- eles são executados de forma autônoma. Assim, nas linhas 8-9, o observável emite duas notificações (onNext, onCompleted) antes que o observador receba a notificação [onNext] (linha 10);
A biblioteca RxJava é responsável pela transferência de dados (as emissões) da thread do observável para a thread do observador. O desenvolvedor não precisa se preocupar com isso.
Vimos como criar observáveis (Observable.from, Observable.create). Agora, vamos ver os observáveis predefinidos da biblioteca RxJava.
7.3. Observáveis predefinidos
7.3.1. Exemplo-08: o método [Observable.range]
 |  |
A partir de agora, utilizaremos classes dedicadas para os processos observados e seus observadores. A ideia é poder registrar seus nomes, suas threads de execução e os horários de execução, a fim de acompanhá-los ao longo do tempo.
A classe [Process] será simplesmente um Observable que pode ser nomeado. Ela implementará a seguinte interface [IProcess]:
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcess<T> {
// nome do observável
public String getName();
// observável
public Observable<T> getObservable();
}
Essa interface poderá ser implementada pela seguinte classe [Process<T>]:
package dvp.rxjava.observables.utils;
import rx.Observable;
import rx.Scheduler;
public class Process<T> implements IProcess<T>{
// nome da variável observável
protected String name;
// processo observado
protected Observable<T> observable;
// construtores
public Process(String name, Observable<T> observable) {
// inicializações locais
this.name = name;
this.observable = observable;
}
// getters e setters
public String getName() {
return name;
}
public Observable<T> getObservable() {
return observable;
}
}
- linha 9: o nome do processo;
- linha 11: a observável observada;
- linhas 14-18: o construtor;
O observador será descrito pela seguinte classe [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> {
...
}
- na linha 11, a classe Observateur<T> estende a classe Subscriber<T>, que apresentamos brevemente no parágrafo 7.1.3. Vamos utilizá-la como argumento do método [Observable.subscribe]:
// execução observável (observação)
obs1.subscribe(observateur);
O método [Observable.subscribe] utilizado na linha 2 acima tem a seguinte definição:
 |
A função do [Subscriber] é, principalmente, gerenciar os elementos emitidos pelo observável ao qual ele está inscrito por meio dos métodos da interface [Observer]: onNext, onError, onCompleted. A classe [Subscriber] possui os seguintes métodos:
 |
No código da classe [Observateur], utilizaremos o método [1] isUnsubscribed para verificar se a assinatura do assinante foi cancelada ou não. A classe [Observateur<T>] completa é a seguinte:
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> {
// um semáforo
private CountDownLatch latch;
// um método de exibição
private Consumer<String> showInfos;
// o nome do observador
private String observerName;
// o nome do processo observado
private String processName;
// construtores
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;
}
// --------------------------- implementação da interface Observer<T>
@Override
public void onCompleted() {
// fim das transmissões
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s,%s].onCompleted", observerName, processName));
}
// fim do bloqueio da thread principal
latch.countDown();
}
@Override
public void onError(Throwable e) {
// erro de emissão
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber [%s, %s].onError (%s)", observerName, processName, e));
}
}
@Override
public void onNext(T value) {
// uma emissão 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));
}
}
}
}
- Além das características de um Subscriber, o observador Observateur incluirá as seguintes informações:
- linha 14: um guard-barrier ou semáforo que servirá para bloquear o thread principal até que o observador tenha recebido todos os elementos emitidos pelo observável. Isso ocorrerá na linha 36 do código, quando o observador receber do observável a notificação de fim de emissão;
- linha 16: uma instância Consumer<String> que servirá para exibir uma mensagem no console;
- linha 18: o nome do observador para diferenciá-los uns dos outros quando houver vários;
- linha 20: o nome do processo observado;
- linhas 36, 46, 54: os métodos [onCompleted, onError, onNext] da interface [Observer<T>] implementada pela classe abstrata [Subscriber<T>]. Essa classe não os implementa. Portanto, isso deve ser feito em suas classes filhas. Antes de realizar qualquer ação nesses métodos, verifica-se se o observador não foi desabonado do observável que ele observa;
- linha 59: o método [onNext] do observador grava a string jSON do elemento recebido. Isso nos permitirá exibir diversos tipos de elementos;
Dito isso, vamos examinar um novo método da classe Observable, o método [range]:
 |
O observável Observable.range(n,m) emite (m) inteiros que variam de n a n+m-1. Vamos analisá-lo com o código [Exemple08] a seguir:
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);
// configuração observável
Observable<Integer> obs1 = Observable.range(15, 3).subscribeOn(Schedulers.computation());
// execução observável (observação)
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();
// fim
showInfos.accept("main : fin observation");
}
// exibições
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()));
}
- linha 16: vamos utilizar dois observadores;
- linha 19: o semáforo é inicializado com o valor 2, pois colocaremos cada observador em um thread diferente. O thread principal deverá, portanto, aguardar a conclusão dos dois threads de observação;
- linha 22: configuramos o observável de forma que ele seja executado em uma thread do agendador [Schedulers.computation()]. O observador estará na mesma thread que o observável;
- linhas 25-27: inscrevemos dois observadores no observável. Isso acionará a execução completa do observável para cada um dos observadores: os números inteiros 15, 16 e 17 serão emitidos;
- linha 30: o thread principal aguarda o término dos observadores;
Os resultados obtidos são os seguintes:
- linha 2: o thread principal está bloqueado, aguardando a conclusão dos dois observadores;
- linhas 3-4: vemos que o observador 0 está na thread [RxComputationThreadPool-1] e o observador 1 na thread [RxComputationThreadPool-2];
- linhas 3-10: vemos que os dois observadores recebem exatamente os mesmos elementos;
Vamos utilizar a classe Observateur, assim definida, para ilustrar o comportamento de outros tipos de observáveis.
7.3.2. Exemplo-09: os métodos Observable.[interval, take, doNext]
 |
 |
Este exemplo ilustra o uso do observável Observable.interval (intervalo longo, unidade TimeUnit), que emite inteiros longos em intervalos regulares. É importante observar o ponto [1]: por padrão, o observável [Observable.interval] é executado em uma das threads do agendador [Schedulers.computation].
O código será o seguinte:
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);
// configuração observável
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// execução observável (observação)
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();
// fim
showInfos.accept("main : fin observation");
}
// visualizações
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()));
}
- linha 22: o observável emite inteiros longos a cada 500 milissegundos. A série começa com o número 0;
- linha 22: esse observável emite um número infinito de valores. O método [Observable.take(n)] cria um novo observável que retém apenas os primeiros n elementos emitidos;
 |
Voltemos ao código do observável:
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
Na linha 2, o método [Observable.doOnNext] é executado sempre que o observável emite um novo elemento. Isso é frequentemente usado para registrar informações. Aqui, queremos registrar a data de emissão dos elementos para verificar se o intervalo de 500 milissegundos está sendo respeitado. O método [Observable.doOnNext] não altera o observável ao qual se aplica. Sua definição é a seguinte:
 |
A execução produz os seguintes resultados:
- linhas 3, 7 e 11: observa-se que o intervalo de emissão é aproximadamente próximo a 500 ms;
- Os dois observadores estão, obviamente, em dois threads diferentes, embora o observável não tenha sido configurado para ser executado com um agendador específico. Esse é o funcionamento padrão do observável [Observable.interval] que vemos aqui;
7.3.3. Exemplos-10/12: os métodos Observable.[error, empty, never]
 |  |
A partir de agora, seremos mais concisos em nossas ilustrações dos métodos da classe [Observable]. O código anterior era o seguinte:
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);
// configuração observável
Observable<Long> obs1 = Observable.interval(500L, TimeUnit.MILLISECONDS).take(3)
.doOnNext(l -> showInfos.accept(l.toString()));
// execução observável (observação)
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();
// fim
showInfos.accept("main : fin observation");
}
// exibições
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()));
}
Esse código já havia sido utilizado no exemplo anterior. Apenas as linhas 21 e 22 mudavam. Portanto, vamos fatorar a maior parte desse código na seguinte classe [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);
// execução observável (observação)
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();
// fim
showInfos.accept("main : fin observation");
}
// visualizações
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()));
}
- linha 13: o método aceita dois parâmetros:
- nbObservateurs: o número de observadores dos processos passados como segundo parâmetro;
- processes: os processos (observáveis nomeados) a serem observados. Graças à notação [IProcess<?>], os processos poderão emitir elementos de tipos diferentes;
- linha 16: o semáforo deve passar para verde quando todos os observadores tiverem concluído todas as suas observações. O valor inicial do semáforo é, portanto, o número de observadores multiplicado pelo número de observações;
- linhas 20-25: cada observador é inscrito em todos os processos que deve observar;
- linha 23: recupera-se o observável a partir do processo (ver parágrafo 7.3.1);
- linha 23: inscreve-se um observador nele. São passadas a ele quatro informações:
- seu nome;
- o semáforo que ele deve decrementar ao receber a notificação de fim de emissão da variável observável que está observando;
- o método a ser utilizado quando ele quiser registrar informações no console;
- o nome do processo que ele irá observar;
Com essas classes definidas, o exemplo 10 será o seguinte:
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 {
// configuração observável
Observable<?> obs = Observable.error(new RuntimeException("Erreur !!!")).subscribeOn(Schedulers.computation());
// execução (observação) observável
ProcessUtils.subscribe(2,new Process<>("process1", obs));
}
}
Na linha 11, o método estático [Observable.error] é definido da seguinte forma:
 |
A linha 8, portanto, configura um observável que se limita a lançar uma exceção para o método [onError] de seus assinantes. A execução produz os seguintes 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]
Nas linhas 3 e 4, o método [onError] dos dois assinantes recebeu a exceção lançada pelo observável.
Essa execução apresenta uma particularidade: os métodos [onCompleted] dos dois observadores não foram chamados. Consequentemente, a barreira não foi abaixada e a thread principal permanece bloqueada no método estático [ProcessUtils.subscribe] na linha 3 a seguir:
// em espera
showInfos.accept("main : attente fin observation");
latch.await();
// fim
showInfos.accept("main : fin observation");
Percebemos aqui que, em caso de erro do observável, o método [onCompleted] dos assinantes não é chamado. Modificamos, então, o método [Observateur.onError] da seguinte maneira:
@Override
public void onError(Throwable e) {
// erro de transmissão
if (!isUnsubscribed()) {
showInfos.accept(String.format("Subscriber[%s, %s].onError (%s)", observerName, processName, e));
}
// fim do bloqueio da thread principal
latch.countDown();
}
Adicionamos as linhas 7 e 8 para remover a restrição em caso de erro do observável. Com esse novo código, a execução produz os seguintes 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]
Obtemos a linha 5, que não tínhamos obtido anteriormente.
O exemplo 11 será o seguinte:
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 {
// configuração observável
Observable<?> obs1 = Observable.empty();
// execução (observação) observável
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
Na linha 10, o método estático [Observable.empty] cria um observável que não emite nenhum elemento. Ele emite apenas a notificação de fim de emissão;
 |
A execução do código do exemplo acima produz os seguintes resultados:
- linhas 2 e 3: percebe-se que os dois observadores recebem a notificação de fim de emissão sem terem recebido nenhum elemento anteriormente.
Pode-se questionar para que serve esse método. Ele pode ser utilizado de forma análoga a uma coleção, inicialmente vazia, na qual se acumulam elementos posteriormente:
Na linha 3, funde-se o observável inicial obs (linha 1) com outros observáveis.
O exemplo 12 ilustra o 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 {
// configuração observável
Observable<?> obs1 = Observable.never();
// execução (observação) observável
ProcessUtils.subscribe(2,new Process<>("process1",obs1));
}
}
O método estático [Observable.never] cria um observável que nunca emite:
 |
A execução do exemplo produz os seguintes resultados:
Na linha 2, o thread principal fica aguardando indefinidamente. De fato, nenhum observável emite a notificação [onCompleted] que permite que o semáforo (barreira) passe para verde (abaixar a barreira).
7.4. Multi-threading
7.4.1. Exemplo 13: thread de ação, thread de observação
No parágrafo 7.1.3, criamos um observável com o método estático [Observable.create]:
|
- o método [create] retorna um tipo Observable<T>;
- o parâmetro do método [create] é uma função do tipo [Observable.OnSubscribe<T>] definida da seguinte forma:
|
O tipo [Observable.OnSubscribe<T>] é uma interface funcional que, por sua vez, estende a interface funcional [Action1<Subscriber<? super T>>]. O método [call] dessa interface espera um tipo [Subscriber] (assinante, subscritor, observador). No restante deste documento, por vezes nos referiremos ao tipo [Observable.OnSubscribe<T>] como uma ação. Vamos criar ações personalizadas que terão um nome. Serão instâncias da seguinte interface [IProcessAction]:
 |
package dvp.rxjava.observables.utils;
import rx.Observable;
public interface IProcessAction<T> extends Observable.OnSubscribe<T> {
// a ação tem um nome
public String getName();
}
- linha 5: a interface [IProcessAction<T>] possui todas as características da interface [Observable.OnSubscribe<T>];
- linha 8: além disso, ela possui um método [getName] que retorna o nome da instância que implementa a interface;
Vamos utilizar a ação denominada [ProcessAction01] a seguir:
package dvp.rxjava.observables.utils;
import java.util.Random;
import rx.Subscriber;
import rx.functions.Func1;
public class ProcessAction01<T> implements IProcessAction<T> {
// dados
private String name;
private int nbValues;
private Func1<Integer, T> func1;
// construtores
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++) {
// expectativa
try {
Thread.sleep(new Random().nextInt(500));
} catch (InterruptedException e) {
// erro
ProcessUtils.showInfos.accept(String.format("Observable (%s) onError", getName()));
subscriber.onError(e);
}
// emissão de um elemento
T value = func1.call(i);
ProcessUtils.showInfos.accept(String.format("Observable (%s,%s) onNext (%s)", getName(), i, value));
subscriber.onNext(value);
}
// concluído
ProcessUtils.showInfos.accept(String.format("Observable (%s) onCompleted", getName()));
subscriber.onCompleted();
}
@Override
public String getName() {
return name;
}
}
- linha 8: a classe [ProcessAction01<T>] implementa a interface [IProcessAction<T>] e, portanto, a interface [Observable.OnSubscribe<T>];
- linha 11: o nome da ação;
- linha 12: o número de valores a serem emitidos;
- linha 13: uma instância do tipo [Func1<Integer, T>] que, a partir de um inteiro, cria um tipo T que será emitido pelo observável (linhas 35 e 37);
- linhas 16-20: passam-se ao construtor o nome da ação, o número de valores a serem emitidos e a função de emissão;
- linhas 23-42: o código do processo;
- linha 23: o método [call] recebe como parâmetro o assinante do observável associado ao processo;
- linha 28: o processo emite seus elementos após uma espera de duração aleatória;
- linha 32: a emissão de um erro;
- linha 37: uma emissão normal;
- linha 41: emissão da notificação de fim de emissão;
- linhas 25-38: a ação emite valores reais de nbValues após um tempo de espera aleatório (linha 30);
- linha 35: o valor a ser emitido é fornecido pela função [func1] passada como parâmetro ao construtor (linha 16);
Reestruturamos a classe [Process] (ver parágrafo 7.3.1) para que ela também possa ser instanciada com uma ação nomeada. Adicionamos a ela o seguinte construtor:
public Process(IProcessAction<T> na, Scheduler schedulerObserved, Scheduler schedulerObserver) {
// nome do processo=nome da ação
name = na.getName();
// ação --> observável
observable = Observable.create(na);
// thread de execução do processo observado
if (schedulerObserved != null) {
observable = observable.subscribeOn(schedulerObserved);
}
// thread de observação do observador
if (schedulerObserver != null) {
observable = observable.observeOn(schedulerObserver);
}
}
- na linha 1, o construtor aceita 3 parâmetros:
- a ação nomeada que será usada para construir o observável (linha 5);
- o agendador do processo observado (pode ser null);
- o agendador do observador (pode ser null);
- linha 5: o observável é criado a partir da ação passada como parâmetro;
O código a seguir, [Exemple13], observa diferentes observáveis:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 1, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// processo 3
Process<Integer> process3 = new Process<>(new ProcessAction01<Integer>("process3", 3, i -> i * 2), null,
Schedulers.computation());
// processo 4
Process<Boolean> process4 = new Process<>(new ProcessAction01<Boolean>("process4", 4, i -> i % 2 == 0), null, null);
// assinaturas
ProcessUtils.subscribe(1, process1);
ProcessUtils.subscribe(1, process2);
ProcessUtils.subscribe(1, process3);
ProcessUtils.subscribe(1, process4);
}
}
- linhas 13-15: o processo process1 gera 1 número real em uma thread de cálculo, que será observado em outra thread de cálculo;
- linhas 17-18: o processo process2 gera 2 cadeias de caracteres em um thread de cálculo, e não há indicação sobre o thread do observador. Os resultados mostram que a observação ocorre, por padrão, no mesmo thread em que o processo é executado;
- linhas 20-21: o processo process3 gera 3 números inteiros em um thread não especificado, que serão observados em um thread de cálculo. Os resultados mostram que a execução do processo ocorre, por padrão, no thread principal;
- linha 23: o processo process4 gera 4 valores booleanos em um thread não especificado, que serão observados em um thread não especificado. Os resultados mostram que a execução do processo e sua observação ocorrem, por padrão, no thread principal;
O resultado da execução deste código é o seguinte:
- o processo process1 gera 1 número real (linha 4) na thread de cálculo [RxComputationThreadPool-4], que é observado na thread de cálculo [RxComputationThreadPool-3] (linha 6);
- o processo process2 gera 2 cadeias de caracteres (linhas 12, 14) na thread de cálculo [RxComputationThreadPool-5], que são observadas nessa mesma thread (linhas 13, 15);
- o processo process3 gera 3 números inteiros (linhas 21, 23, 25) na thread principal, que são observados na thread de cálculo [RxComputationThreadPool-6] (linhas 22, 24, 28);
- o processo process4 gera 4 valores booleanos (linhas 34, 36, 38, 40) na thread principal, que são observados nessa mesma thread principal (linhas 33, 35, 37, 39);
Convidamos o leitor a acompanhar acima:
- o ciclo de vida do processo observado e de sua thread;
- o ciclo de vida de seu observador e sua thread;
Grande parte do interesse das bibliotecas Rx reside nesse multithreading, que o desenvolvedor não precisa gerenciar por conta própria.
7.5. Combinações de vários observáveis
7.5.1. Exemplo 14: mesclar dois observáveis com [Observable.merge]
Apresentamos agora métodos estáticos da classe [Observable] que permitem combinar vários observáveis em um observável resultante.
O primeiro exemplo desse tipo será o seguinte:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), Schedulers.computation(), null);
// fusão
Process<?> process12 = new Process<>("process12",
Observable.merge(process1.getObservable(), process2.getObservable()));
// assinaturas
ProcessUtils.subscribe(1, process12);
}
}
- linhas 15-17: um processo chamado [process1] emitirá 3 números reais em uma thread de cálculo. Ele também será observado em uma thread de cálculo;
- linhas 19-20: um processo chamado [process2] emitirá 2 cadeias de caracteres em um thread de cálculo. O thread de observação não é imposto. Vimos anteriormente que, nesse caso, o thread de observação é o thread de cálculo;
- linha 23: os dois processos são fundidos, ou seja, cria-se um observável cujos elementos provêm simultaneamente dos dois processos. Para isso, utiliza-se o método estático [Observable.merge]:
 |
Ao contrário do que o esquema acima poderia sugerir, durante a fusão, os elementos de um fluxo 1 podem ser intercalados entre os elementos de um fluxo 2. É o que mostram os resultados da execução:
- linha 3: o processo [process1] é executado na thread de cálculo [RxComputationThreadPool-4];
- linha 4: o processo [process2] é executado na thread de cálculo [RxComputationThreadPool-5];
- linha 9: o processo [process12] é observado na thread de cálculo [RxComputationThreadPool-3]. Não sei qual foi a regra que levou a essa escolha;
- linhas 9-11: percebe-se que o observador observa elementos dos dois processos [process1] (linha 5) e [process2] (linhas 6, 7), embora nenhum dos dois tenha sido concluído (há mistura);
- o processo [process12] é encerrado (linha 17) quando os dois processos process1 e process2 são encerrados;
7.5.2. Exemplo 15: concatenar dois observáveis com [Observable.concat]
Vamos agora examinar o código a seguir:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 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()));
// assinaturas
ProcessUtils.subscribe(1, process12);
}
}
- linhas 15-17: um processo chamado [process1] emitirá 3 números reais em uma thread de cálculo. Ele também será observado em uma thread de cálculo;
- linhas 19-20: um processo chamado [process2] emitirá duas cadeias de caracteres em uma thread não especificada, neste caso a thread principal por padrão. Ele será observado em uma thread de cálculo;
- linha 23: os dois processos são concatenados, ou seja, cria-se um observável cujos elementos provêm dos dois processos. Não há mistura dos valores emitidos. O processo [process12] emitirá primeiro todos os valores do processo [process1] e, em seguida, os do processo [process2]. Para isso, utiliza-se o método estático [Observable.concat]:
 |
Os resultados da execução são os seguintes:
- linhas 3-10: o processo [process1] é executado e o processo [process12] emite os valores gerados pelo [process1];
- linha 9: o processo [process1] foi concluído;
- linhas 11-17: o processo [process2] é executado e o processo [process12] transmite os valores emitidos pelo [process2];
Há uma peculiaridade no processo process2: não havia sido definido um thread de execução. Seria de se esperar, portanto, que, por padrão, ele fosse o thread principal. No entanto, não é o caso. A thread de execução foi a thread de cálculo [RxComputationThreadPool-3] (linha 11). Portanto, quando não se define uma thread de execução ou de observação, não é possível fazer suposições sobre qual thread será escolhida.
7.5.3. Exemplo 16: combinar duas variáveis observáveis com [Observable.zip]
Vamos agora examinar o código a seguir:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<String> process2 = new Process<>(
new ProcessAction01<String>("process2", 2, i -> String.format("valeur-%s", i)), null, null);
// função de combinação dos dois processos
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");
}
}
};
// compactação dos 2 processos
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
// assinaturas
ProcessUtils.subscribe(1, process12);
}
}
- linhas 16-18: um processo chamado [process1] emitirá 3 números reais em uma thread de cálculo. Ele também será observado em uma thread de cálculo;
- linhas 20-21: um processo chamado [process2] emitirá 2 cadeias de caracteres em um thread não imposto. O thread de observação também não é imposto;
- linhas 23-32: instanciação de um tipo [FuncN<String>] com uma classe anônima. FuncN é uma interface funcional:
 |
O método [FuncN.call] espera um array de objetos e retorna um tipo R. A função [funcn] será utilizada para combinar os processos process1 e process2 nessa ordem. No método [FuncN.call]:
- args[0] será um Double;
- args[1] será um String;
Aqui, o resultado de [funcn.call] será a sequência de caracteres da linha 27. A construção desse resultado não requer o conhecimento dos tipos dos argumentos do método call.
Os dois processos são combinados da seguinte maneira:
// compactação dos 2 processos
Process<String> process12 = new Process<>("process12",
Observable.zip(Arrays.asList(process1.getObservable(), process2.getObservable()), funcn));
O método [Observable.zip] funciona da seguinte maneira:
 |
Observa-se que:
- o primeiro argumento do zip é um Iterable<Observable>. No nosso exemplo, temos um parâmetro efetivo do tipo List<Observable> formado por nossos dois observáveis;
- o segundo argumento do zip é do tipo FuncN. No nosso exemplo, o parâmetro efetivo é [funcn];
A execução produz os seguintes resultados:
- linhas 7, 11: o processo process12 emite dois elementos;
- linha 8: o elemento adicional emitido pelo processo process1, que não possui parceiro no processo process2, não é emitido pelo processo de resultado process12;
Percebe-se que o processo process2, ao qual não havia sido atribuído nem thread de execução nem thread de observação, utilizou o thread principal para ambos.
7.5.4. Exemplo 17: combinar dois observáveis com [Observable.combineLatest]
Vamos agora examinar o código a seguir:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null,
Schedulers.computation());
// combinação dos dois processos
Process<Double> process12 = new Process<>("process12",
Observable.combineLatest(process1.getObservable(), process2.getObservable(), (d1, d2) -> d1 + d2));
// assinaturas
ProcessUtils.subscribe(1, process12);
}
}
- linhas 14-16: um processo chamado [process1] emitirá 3 números reais em uma thread de cálculo. Ele também será observado em uma thread de cálculo;
- linhas 18-20: um processo chamado [process2] emitirá 2 números reais em um thread não restrito. Eles serão observados em um thread de cálculo;
- linha 23: os dois observáveis são combinados com o seguinte método estático [Observable.combineLatest]:
 |
O observável [combineLatest] funciona da seguinte maneira: quando um dos dois observáveis emite um elemento E1, esse elemento é combinado por [combineFunction] com o último elemento emitido pelo outro observável.
A execução deste código produz o seguinte resultado:
- linha 5: a emissão de process2 (56) é combinada com o último elemento emitido por process1 (54, linha 4) e produz o resultado da linha 7;
- linha 6: a emissão de process1 (51,6) é combinada com o último elemento emitido por process2 (56, linha 5) e produz o resultado da linha 8;
- linha 9: a transmissão de process2 (261,8) é combinada com o último elemento transmitido por process1 (51,6, linha 6) e produz o resultado da linha 12;
- linha 13: a emissão de process1 (80,39) é combinada com o último elemento emitido por process2 (261,8, linha 9) e produz o resultado da linha 15;
Estamos aqui diante de uma variante do observável [zip], em que, desta vez, os elementos combinados não são necessariamente os elementos que ocupam a mesma posição nos fluxos. Observa-se aqui que o processo process2, ao qual não havia sido atribuída uma thread de execução, foi executado na thread principal (linha 2).
7.5.5. Exemplo 18: combinando dois observáveis com [Observable.amb]
Vamos agora examinar o código a seguir:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Double> process2 = new Process<>(
new ProcessAction01<Double>("process2", 2, i -> new Random().nextInt(200) * 1.4), null, null);
// combinação dos dois processos
Process<Double> process12 = new Process<>("process12",
Observable.amb(process1.getObservable(), process2.getObservable()));
// assinaturas
ProcessUtils.subscribe(1, process12);
}
}
- linhas 14-16: um processo chamado [process1] emitirá 3 números reais em uma thread de cálculo. Ele também será observado em uma thread de cálculo;
- linhas 18-20: um processo chamado [process2] emitirá 2 números reais em um thread não restrito. Eles serão observados em um thread não restrito;
- linha 22: os dois observáveis são combinados com o seguinte método estático [Observable.amb]:
 |
Conforme mostra o esquema acima, o observável [Observable.amb(Observable o1, Observable o2)] emite os elementos do observável que os emite primeiro. Isso é confirmado pelos resultados do exemplo apresentado:
- linha 4: é o processo process2 que emite primeiro;
- linhas 8 e 12: o processo process12 emite todos os elementos emitidos pelo processo process2 (linhas 4 e 11);
7.6. Cadeia de processamento de um observável
7.6.1. Exemplo 19: transformar um observável com [Observable.map]
Nos exemplos anteriores, examinamos diversas combinações de dois observáveis para formar um terceiro observável. Apresentamos agora métodos estáticos da classe [Observable] que permitem operações de transformação, filtragem e agregação em um observável. Encontraremos aqui métodos análogos aos da classe [Stream] estudados no parágrafo 5.
Nosso primeiro exemplo será o seguinte:
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 {
// processo 1
Process<Double> process1 = new Process<>(
new ProcessAction01<Double>("process1", 3, i -> new Random().nextInt(100) * 1.2), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<String> process2 = new Process<>("process2",
process1.getObservable().map(d -> String.format("valeur-%s", d)));
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
- linhas 14-16: um processo chamado process1 emitirá três números reais em uma thread de cálculo. Ele também será observado em uma thread de cálculo;
- linhas 17-18: os números gerados por process1 serão transformados em cadeias de caracteres no processo process2;
- linha 20: observa-se process2;
O método [Observable.map] da linha 18 é análogo ao método [Stream.map] estudado no parágrafo 5.5:
 |
Os resultados do exemplo são os seguintes:
- linhas 4, 5 e 8: as emissões de process1. São números reais;
- linhas 6, 7 e 10: as emissões de process2 observadas. São cadeias de caracteres;
7.6.2. Exemplo-20: filtrar uma variável observável com [Observable.filter]
O exemplo será o seguinte:
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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().filter(i -> i % 2 == 0));
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
- linhas 11-12: um processo chamado process1 emitirá os números inteiros de 0 a 2 em um thread de cálculo. Ele também será observado em um thread de cálculo;
- linha 14: os números gerados por process1 serão filtrados para que apenas os números pares sejam mantidos em process2;
- linha 20: observa-se process2;
O método [Observable.filter] da linha 18 é análogo ao método [Stream.filter] estudado no parágrafo 5.4:
 |
Os resultados do exemplo são os seguintes:
- linhas 4, 5 e 7: as transmissões de process1;
- linhas 6 e 9: as emissões de process2 observadas. São os elementos de process1 que são pares;
7.6.3. Exemplo 21: transformar um observável com [Observable.flatMap]
O exemplo será o seguinte:
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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Integer> process2 = new Process<>("process2", process1.getObservable().flatMap(i -> {
int value = i * 10;
return Observable.just(value, value + 1, value + 2);
}));
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
- linhas 12-13: um processo chamado process1 emitirá os números inteiros de 0 a 2 em um thread de cálculo. Ele também será observado em um thread de cálculo;
- linhas 15-18: cada número n gerado por process1 é transformado em um observável que gera os 3 números (10*n, 10*n+1, 10*n+2). Se, na linha 15, fosse utilizado o método [map], o process2 emitiria um tipo Observable<Integer> e não um tipo Integer. O método [flatMap] utilizado permite simplificar (flatten) essa sequência de elementos do tipo Observable<Integer> em uma sequência de elementos do tipo Integer, constituída por cada um dos elementos de cada um dos Observable<Integer>;
- linha 20: observa-se process2;
O método [Observable.flatMap] da linha 15 é análogo ao método [Stream.flatMap] analisado no parágrafo 5.6.12:
 |
Os resultados do exemplo são os seguintes:
- linhas 5-7: as três transmissões de process2 após a transmissão da linha 4 de process1;
- linhas 9-11: as três emissões de process2 após a emissão da linha 8 de process1;
- linhas 14-16: as três emissões de process2 na sequência da emissão da linha 12 de process1;
O código a seguir mostra como criar um tipo Observable<Integer[]> a partir de process1 e [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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Integer[]> process2 = new Process<>("process2", process1.getObservable().map(i -> {
int value = i * 10;
return new Integer[] { value, value + 1, value + 2 };
}));
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
- linha 14: utiliza-se o método [Observable.map];
- linha 16: que retorna um tipo Integer[];
Os resultados são os seguintes:
- linhas 6, 7, 10: vemos os resultados do map;
Todas essas transformações de observáveis podem ser encadeadas, já que cada transformação produz um novo observável. É o que mostra o exemplo a seguir [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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 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));
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
- linhas 15-18: o flatMap é seguido por um filter;
Os resultados da execução são os seguintes:
- linhas 8-13: o process2 emitiu apenas os elementos pares provenientes do flatMap;
Um método semelhante ao [flatMap] é o método [flatMapIterable], ilustrado pelo exemplo a seguir, [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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 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));
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
Na linha 16, em vez de usar o método [flatMap], utiliza-se o método [flatMapIterable]. Nesse caso, a função de transformação deve produzir um tipo Iterable<T> (linha 18) em vez de um tipo Observable<T>.
Obtêm-se os mesmos resultados que anteriormente.
Voltemos à definição do método [flatMap]:
 |
Vemos acima que um elemento azul [3] foi inserido entre os dois elementos verdes [1-2]. Isso significa que, em sua operação de nivelamento dos Observable<T>, o método [flatMap] respeita a ordem de emissão desses diferentes observáveis internos. Isso é demonstrado pelo exemplo a seguir, [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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// processo 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().flatMap(i -> process2.getObservable()));
// assinaturas
ProcessUtils.subscribe(1, process3);
}
}
- linhas 11-12: o processo process1 gera os números inteiros [0,1];
- linhas 14-15: o processo process2 emite os números inteiros [10,11,12];
- linhas 17-18: a cada elemento emitido por process1, é associada a observável do processo process2. Isso significa que:
- ao elemento [0] do process1 será associada uma observável que emite os [10,11,12];
- o mesmo vale para o elemento 1;
No final, os 6 números [10, 11, 12, 10, 11, 12] serão emitidos. Queremos verificar em que ordem.
Os resultados da execução são os seguintes:
Vemos que a ordem de emissão do processo process3 foi: [10, 10, 11, 12, 11, 12] (linhas 11, 12, 14, 17, 19, 22). Portanto, houve de fato uma mistura dos elementos emitidos pelo processo process2. É possível evitar isso utilizando o método [concatMap] em vez do método [flatMap]. É o que mostra o código a seguir, [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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// processo 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().concatMap(i -> process2.getObservable()));
// assinaturas
ProcessUtils.subscribe(1, process3);
}
}
Na linha 18, substituímos [flatMap] por [concatMap]. Os resultados da execução são os seguintes:
Observa-se que a ordem de emissão do processo process3 foi: [10, 11, 12, 10, 11, 12] (linhas 12-14, 17, 19, 22). Os elementos emitidos pelo processo process2 não foram misturados.
Outra variante do método [map] é o método [switchMap]:
 |
Acima, a partir do observável [1], surgem outros 3 observáveis [2] com 2 elementos, que são então achatados como em [flatMap] e [3]. Pode-se observar que o resultado tem 5 elementos e não 6. Isso se deve ao fato de que, antes que o segundo observável emita seu elemento nº 2, [6], o terceiro observável já emitiu seu primeiro elemento, [5], o que faz com que o segundo observável seja descartado. Portanto, o elemento [6] não aparece no observável resultante [3].
Para ilustrar o [switchMap], utilizaremos o seguinte exemplo [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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 2, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Integer> process2 = new Process<>(new ProcessAction01<>("process2", 3, i -> i + 10),
Schedulers.computation(), Schedulers.computation());
// processo 3
Process<Integer> process3 = new Process<>("process3",
process1.getObservable().switchMap(i -> process2.getObservable()));
// assinaturas
ProcessUtils.subscribe(1, process3);
}
}
A execução do exemplo produz os seguintes resultados:
- process1 emite 2 elementos que dão origem a 2 observáveis process2 de 3 elementos;
- linha 14: o observador recebe o elemento nº 0 emitido pelo primeiro observável process2 na linha 6;
- linha 15: o observador recebe o elemento nº 0 emitido pelo segundo observável process2, na linha 13. Não se sabe por que ele não recebeu anteriormente os elementos 1 e 2 emitidos pelo primeiro observável process2 nas linhas 7 e 8. De qualquer forma, o primeiro observável process2 é descartado;
- no final, o observador vê apenas 4 elementos (linhas 14, 15, 17, 20) em vez dos 6 que foram emitidos;
7.6.4. Exemplos-22: outros métodos da classe [Observable]
A classe [Observable] retoma vários métodos da classe [Stream] com um funcionamento análogo. Aqui estão alguns deles. Limitar-nos-emos a apresentar o código e seus 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 {
// processo
Process<Integer> process = new Process<>("process", Observable.range(1, 10).take(3));
// assinaturas
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 {
// processo
Process<Integer> process = new Process<>("process", Observable.range(1, 10).takeLast(2));
// assinaturas
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 {
// processos
Process<Integer> process = new Process<>("process", Observable.range(1, 10).skip(5).take(2));
// assinaturas
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 {
// processos
Process<Integer> process = new Process<>("process", Observable.range(1, 10).reduce(0, (i, a) -> i + a));
// assinaturas
ProcessUtils.subscribe(1, process);
}
}
- linha 10: calcula a soma dos elementos do observável. O resultado é um observável que emite essa soma;
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 {
// processos
Process<Boolean> process = new Process<>("process", Observable.range(1, 10).all(i -> i > 10));
// assinaturas
ProcessUtils.subscribe(1, process);
}
}
- linha 10: retorna um Observable<Boolean> que emite o elemento true, se o predicado do método [all] for verdadeiro para todos os elementos; caso contrário, retorna 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 {
// processos
Process<Integer> process = new Process<>("process", Observable.range(1, 10).count());
// assinaturas
ProcessUtils.subscribe(1, process);
}
}
- linha 10: [Observable.count] cria um observável com 1 elemento, que é a soma dos 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 {
// processos
Process<Integer> process = new Process<>("process", Observable.just(1, 2, 1, 3).distinct());
// assinaturas
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 {
// processos
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()));
// assinaturas
ProcessUtils.subscribe(1, process);
}
}
- linha 11: o método [groupBy] agrupa os 10 elementos emitidos em 2 grupos: os números pares e os números ímpares. O resultado é um tipo Observable<GroupedObservable<Boolean, Integer>>, ou seja, um observável cujos elementos são do tipo GroupedObservable<Boolean, Integer>, onde Boolean é o tipo da chave do grupo (false, true neste caso) e que também é o tipo do resultado da função lambda passada como parâmetro ao método [groupBy], e Integer o tipo dos elementos do grupo;
- linha 12: o tipo GroupedObservable possui um método [asObservable] que permite criar um observável a partir desse tipo. Teremos, portanto, dois tipos Observable<Integer>: um para os números pares e outro para os números ímpares. A partir desses dois observáveis, o método [concatMap] criará um único observável;
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 {
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
Schedulers.computation());
// processo 2
Process<Timestamped<Integer>> process2 = new Process<>("process2", process1.getObservable().timestamp());
// assinaturas
ProcessUtils.subscribe(1, process2);
}
}
- na linha 15, o método [timestamp] associa uma hora a cada elemento da variável observável processada;
resultados
Neste exemplo, é difícil dizer o que representa a informação timestamp:
- linhas 4-5: percebe-se que o elemento 1 de process1 foi emitido 139 ms após o elemento 0;
- linhas 6 e 7: percebe-se que o elemento 1 de process2 foi observado 234 ms após o elemento 0;
- linhas 5 e 8: percebe-se que o elemento 2 de process1 foi emitido 33 ms após o elemento 1;
- linhas 7 e 10: percebe-se que o elemento 2 de process2 foi observado 37 ms após o elemento 1;
Esses deslocamentos se devem ao fato de que as threads de observação e de execução dos observáveis não são as mesmas. Se substituirmos as linhas 12-13 pelas seguintes (Exemplo22j):
// processo 1
Process<Integer> process1 = new Process<>(new ProcessAction01<>("process1", 3, i -> i), Schedulers.computation(),
null);
- linhas 2-3: não se impõe o thread de observação. Sabe-se que, nesse caso, o observável é observado onde é executado;
Isso resulta nos seguintes resultados:
- linhas 4 e 6: o processo process1 emite seu elemento nº 1 587 ms após seu elemento nº 0;
- linhas 5 e 7: o observador observa esses dois elementos com um intervalo de 586 ms;
- linhas 6 e 8: o processo process1 emite seu elemento nº 2 396 ms após seu elemento nº 1;
- linhas 7 e 9: o observador observa esses dois elementos com um intervalo de 396 ms;
Aqui, os valores do timestamp são consistentes: representam corretamente a data de emissão do elemento.
7.7. Os agendadores
7.7.1. Exemplo-23: o agendador [Schedulers.computation]
Vamos agora examinar os agendadores de execução. A observação será feita no thread de execução.
O tema dos agendadores é um pouco obscuro. Os diferentes agendadores são apresentados nesta questão no site de StackOverflow [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
Vamos tentar ilustrar o uso desses diferentes agendadores por meio de exemplos. O primeiro ilustra o agendador [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 {
// processos
@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);
}
// assinaturas
ProcessUtils.subscribe(1, processes);
}
}
- linhas 14-19: cria-se uma matriz de 10 processos em execução em um thread de cálculo;
- linha 17: cada processo gera um número real aleatório;
- linha 21: assinamos todos esses processos;
Os resultados são os seguintes:
- linhas 2-10: os 8 primeiros processos são iniciados em 8 threads diferentes (a máquina utilizada possui 8 núcleos). É possível observar que todos eles são iniciados aproximadamente ao mesmo tempo;
- linhas 17-19: três processos são encerrados, liberando assim três threads;
- linhas 23-24: os dois últimos processos podem então iniciar, ocupando 2 das threads assim liberadas;
Portanto, deve-se observar que o agendador [Schedulers.computation] fornece um conjunto de n threads, onde n é o número de núcleos da máquina. As threads são executadas em paralelo nesses núcleos.
7.7.2. Exemplo 24: o agendador [Schedulers.io]
Executamos o código anterior com o agendador [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 {
// processos
@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);
}
// assinaturas
ProcessUtils.subscribe(1, processes);
}
}
- linha 18: os processos são executados com as threads do agendador [Schedulers.io];
Isso gera os seguintes resultados:
- linhas 2-10: os 10 processos são iniciados, cada um em um thread diferente. Ao contrário do caso anterior, todos os processos puderam ser iniciados. Observa-se que esses inícios ocorrem em um intervalo de 6 ms, enquanto anteriormente isso levava 1 ms;
- linhas 13-18: os observáveis emitem um após o outro e não de forma quase paralela, como havia ocorrido anteriormente;
Qual é a diferença entre os agendadores [Schedulers.io] e [Schedulers.computation]? Uma resposta pode ser encontrada no URL [http://stackoverflow.com/questions/31276164/rxjava-schedulers-use-cases]:
 |
7.7.3. Exemplo 25: o agendador [Schedulers.newThread]
Executamos o código anterior com o agendador [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 {
// processos
@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);
}
// assinaturas
ProcessUtils.subscribe(1, processes);
}
}
Os resultados obtidos são os mesmos que com o agendador [Schedulers.io]:
Nos exemplos URL e [http://stackoverflow.com/questions/33415881/retrofit-with-rxjava-schedulers-newthread-vs-schedulers-io], explica-se que o agendador [Schedulers.io] fornece um pool de threads, o que não ocorre com o agendador [Schedulers.newThread]. Um pool de threads cria automaticamente um número n de threads. Ele as aloca aos processos que precisam delas. Quando esses processos são concluídos, suas threads não são excluídas, mas retornam ao pool e podem ser reutilizadas por outro processo. Isso é mais econômico do que criar e excluir threads constantemente. Portanto, pode-se considerar que é preferível utilizar o agendador [Schedulers.io].
7.7.4. Exemplo 26: os agendadores [Schedulers.immediate, Schedulers.trampoline]
Voltemos à explicação dada para esses dois agendadores:
|
A explicação é bastante simples de entender, mas, quando se tenta ilustrá-la, percebe-se que não a compreendemos. Foi o livro [Learning Reactive Programming With Java 8] que me permitiu criar um exemplo que retoma um exemplo encontrado nesse livro, mas o simplifica. É o seguinte:
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 {
// um agendador
Scheduler scheduler = Schedulers.immediate();
// um worker desse agendador
Worker worker = scheduler.createWorker();
// um tipo Action0 a ser executado no worker
Action0 action02 = new Action0() {
@Override
public void call() {
// registro de ação02
ProcessUtils.showInfos.accept("action02");
}
};
// um tipo Action0 a ser executado no worker
Action0 action01 = new Action0() {
@Override
public void call() {
// programa-se uma nova ação no mesmo worker
worker.schedule(action02);
// registro da ação01
ProcessUtils.showInfos.accept("action01");
}
};
// a ação 01 está programada no worker
worker.schedule(action01);
}
// visualizações
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()));
}
- linha 17: um agendador. Será ou [Schedulers.immediate], como aqui, ou [Schedulers.trampoline] posteriormente;
- linha 19: é possível executar ações do tipo Action0 (linhas 21, 20) nos workers do agendador. O método [Scheduler.createWorker] permite criar um worker. O método [Worker.schedule(Action0)] permite que um worker execute um tipo Action0;
- linhas 21-27: uma primeira ação chamada [action02], que será executada (linha 40) pelo worker da linha 19;
- linhas 30-38: uma segunda ação chamada [action01]. Ela tem a particularidade de fazer com que a ação action02 seja executada no mesmo worker que ela (linha 34). É aí que reside a diferença entre [Schedulers.immediate] e [Schedulers.trampoline]:
- se o agendador for [Schedulers.immediate], então, na linha 34, a ação action02 será executada imediatamente (daí o nome do agendador) e a ação action01 em andamento será interrompida. Aí, aparecerá a mensagem da linha 25. Concluída a ação action02, a ação action01 será retomada e aparecerá a mensagem da linha 36;
- se o agendador for [Schedulers.trampoline], então, na linha 34, a ação action02 é colocada em espera. Ela só será executada quando a tarefa em andamento action01 for concluída. Veremos então aparecer a mensagem da linha 36. Concluída a ação action01, a ação action02 será executada e veremos a mensagem da linha 25;
A execução do código acima produz os seguintes resultados:
Se, na linha 17, utilizarmos o agendador [Schedulers.trampoline], obtemos os resultados opostos:
Dito isso, é difícil estabelecer uma relação com os observáveis. Não encontrei nenhum exemplo convincente que pudesse demonstrar a vantagem de executar um observável em uma dessas duas threads. Aqui está, no entanto, um exemplo, mas que não 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 {
// Worker
Worker worker = Schedulers.immediate().createWorker();
// Worker worker = Schedulers.trampoline().createWorker();
// observável 1 no worker
worker.schedule(() -> Observable.range(1, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
// observável 2 no mesmo worker
worker.schedule(() -> Observable.range(100, 2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer i) {
ProcessUtils.showInfos.accept(String.valueOf(i));
}
}));
}
}));
}
}
- linhas 13-14: cria-se um worker a partir de um dos dois agendadores [Schedulers.immediate] e [Schedulers.trampoline];
- linha 16: um primeiro observável obs1 é programado nesse worker para emitir os números [1,2]
- linha 22: sempre que um elemento desse observável obs1 é observado, a observação de um segundo observável obs2 é iniciada no mesmo worker para emitir os números [100,101];
Com o agendador [Schedulers.immediate], obtêm-se os seguintes resultados:
Já com o agendador [Schedulers.trampoline], obtêm-se os seguintes resultados:
7.8. Conclusion
Ainda há muito a ser feito. Para aprofundar o conhecimento sobre a biblioteca RxJava, o leitor é convidado a continuar seu aprendizado com as referências fornecidas no início deste documento. Apesar de tudo, já temos as bases para utilizar o RxJava nos ambientes Swing e Android. É isso que vamos demonstrar agora.