8. RxJava no ambiente Swing

8.1. Introduction

Aqui, vamos retomar a aplicação Swing apresentada no parágrafo 2.

Para trabalhar com o RxJava em um ambiente Swing, utilizaremos a biblioteca RxSwing, que adiciona ao RxJava classes e interfaces úteis em um ambiente Swing. Para isso, o arquivo Gradle do exemplo Swing é o seguinte:

buildscript {
    repositories {
        mavenCentral()
    }
}
apply plugin: 'java'
jar {
    baseName = 'exemples-01'
    version = '0.0.1-SNAPSHOT'
}
repositories {
    mavenCentral()
}
dependencies {
    compile('io.reactivex:rxswing:0.25.0')
    compile('io.reactivex:rxjava:1.1.3')
    compile('com.fasterxml.jackson.core:jackson-databind:2.7.3')
}
task wrapper(type: Wrapper) {
    gradleVersion = '2.9'
}

• linha 15: a dependência de RxSwing;

Usaremos apenas um único objeto específico de RxSwing: o agendador [SwingScheduler.getInstance()], que executa e observa os observáveis na thread do evento loop do Swing. Vamos utilizá-lo exclusivamente para observar observáveis que estão sendo executados em threads diferentes daquela do evento loop. Vale lembrar a arquitetura do aplicativo de exemplo:

• a camada de serviço assíncrona apresenta métodos que retornam observáveis. Executamos esses observáveis em threads diferentes daquela do loop de eventos. Assim, a interface gráfica não fica congelada. Ela pode reagir às solicitações do usuário. A forma mais óbvia é permitir que o usuário clique em um botão [Annuler] para interromper uma operação assíncrona que esteja demorando demais. Para que isso seja possível, basta que a interface gráfica esteja congelada (frozen);
• a camada Swing pretende utilizar os resultados retornados pelas operações assíncronas e, a partir deles, atualizar a interface gráfica. No entanto, isso só pode ser feito na thread do event loop. Para isso, esses resultados são monitorados no agendador [SwingScheduler.getInstance()];

Assim, no código de gerenciamento de eventos da interface gráfica, a interação com a camada assíncrona [rxService] ocorre da seguinte forma:

Observable obs=rxService.doSomething(...).subscribeOn(Schedulers.computation()).observeOn(SwingScheduler.getInstance()) ;

onde o agendador [Schedulers.computation()] poderá ser substituído por outro agendador, dependendo dos casos de uso.

Recomenda-se ao leitor que releia o parágrafo 2. Agora, ele possui os conhecimentos necessários para compreendê-lo completamente.

8.2. A estrutura do código

O código implementa a seguinte arquitetura:

O projeto do IntelliJ IDEA que implementa essa arquitetura é o seguinte:

• o pacote [rxswing.service] implementa as camadas de serviço síncronas (IService, Service) e assíncronas (IRxService, RxService);
• o pacote [rxswing.ui] implementa a interface Swing;

8.3. Execução do projeto

Para executar o projeto no IntelliJ IDEA, proceda da seguinte forma:

8.4. O serviço síncrono

A camada de serviço síncrona apresenta a seguinte interface [IService]:

package dvp.rxswing.service;

public interface IService {
  // números aleatórios no intervalo [a,b]
  // são gerados n números, sendo que n é, por sua vez, um número aleatório no intervalo [minCount, maxCount]
  // os números são gerados após um intervalo de espera de delay milissegundos,
  // onde [delay] é, por sua vez, um número aleatório no intervalo [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

O tipo [ServiceResponse] da resposta do serviço é o seguinte:

package dvp.rxswing.service;

import java.util.List;

public class ServiceResponse {

  // tempo de espera do serviço
  private int delay;
  // números aleatórios
  private List<Integer> aleas;
  // thread de execução
  private String executedOn;

  // construtores

  public ServiceResponse() {
      // thread de execução
    executedOn = Thread.currentThread().getName();
  }

  public ServiceResponse(int delay, List<Integer> aleas) {
      // construtor local
    this();
    // outras inicializações
    this.delay = delay;
    this.aleas = aleas;
  }

  // getters e setters
...
}

A interface [IService] é implementada pela seguinte classe [Service]:

package dvp.rxswing.service;

import java.util.*;

public class Service implements IService {

  @Override
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay) {
    // números aleatórios no intervalo [a,b]
    // n números são gerados, sendo n próprio um número aleatório no intervalo [minCount, maxCount]
    // os números são gerados após um intervalo de espera de delay milissegundos,
    // onde [delay] é, por sua vez, um número aleatório no intervalo [minDelay, maxDelay]

    // algumas verificações
    List<String> messages = new ArrayList<>();
    int erreur = 0;
    if (a < 0) {
      messages.add("Le nombre a de l'intervalle [a,b] de génération doit être supérieur à 0");
      erreur |= 2;
    }
    if (a >= b) {
      messages.add("Dans l'intervalle [a,b] de génération, on doit avoir a< b");
      erreur |= 4;
    }
    if (minCount < 0) {
      messages.add("Le nombre min de l'intervalle [min,count] du nombre de valeurs générées doit être supérieur à 0");
      erreur |= 16;
    }
    if (minCount > maxCount) {
      messages.add("Dans l'intervalle [min,count] du nombre de valeurs générées, on doit avoir min<= max");
      erreur |= 32;
    }
    if (minDelay < 0) {
      messages.add("Le nombre min de l'intervalle [min,count] du délai d'attente doit être supérieur à 0");
      erreur |= 64;
    }
    if (minCount > maxCount) {
      messages.add("Dans l'intervalle [min,count] du délai d'attente, on doit avoir min<= max");
      erreur |= 128;
    }
    if (maxDelay > 5000) {
      messages.add("L'attente en millisecondes avant la génération des nombres doit être dans l'intervalle [0,5000]");
      erreur |= 256;
    }
    // erros?
    if (!messages.isEmpty()) {
      throw new AleasException(String.join(" [---] ", messages), erreur);
    }
    // gerador de números aleatórios
    Random random = new Random();
    // em espera?
    int delay = minDelay + random.nextInt(maxDelay - minDelay + 1);
    if (delay > 0) {
      try {
        Thread.sleep(delay);
      } catch (InterruptedException e) {
        throw new AleasException(String.format("[%s : %s]", e.getClass().getName(), e.getMessage()), 1024);
      }
    }
    // geração do resultado
    int count = minCount + random.nextInt(maxCount - minCount + 1);
    List<Integer> nombres = new ArrayList<>();
    for (int i = 0; i < count; i++) {
      nombres.add(a + random.nextInt(b - a + 1));
    }
    // retorno do resultado
    return new ServiceResponse(delay,nombres);
  }

}

A classe de exceção [AleasException] utilizada pelo serviço é a seguinte:

package dvp.rxswing.service;

public class AleasException extends RuntimeException {

    private static final long serialVersionUID = 1L;
    // código de erro
  private int code;

  // construtores
  public AleasException() {
  }

  public AleasException(String detailMessage, int code) {
    super(detailMessage);
    this.code = code;
  }

  public AleasException(Throwable throwable, int code) {
    super(throwable);
    this.code = code;
  }

  public AleasException(String detailMessage, Throwable throwable, int code) {
    super(detailMessage, throwable);
    this.code = code;
  }

  // getters e setters
...
}

• linha 3: ela estende a classe [RuntimeException]. Trata-se, portanto, de uma exceção não controlada;
• linha 7: ela adiciona um código de erro à sua classe pai (0 = sem erro);

8.5. O serviço assíncrono

A camada de serviço assíncrona apresenta a seguinte interface [IRxService]:

package dvp.rxswing.service;

import dvp.rxswing.ui.UiResponse;
import rx.Observable;

public interface IRxService {
  // números aleatórios no intervalo [a,b]
  // n números são gerados, sendo n próprio um número aleatório no intervalo [minCount, maxCount]
  // os números são gerados após um intervalo de espera de delay milissegundos,
  // onde [delay] é, por sua vez, um número aleatório no intervalo [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

• linha 11: agora é um observável que o método [getAleas] do serviço retorna;

O método [getAleas] retorna uma resposta do tipo [UiResponse] destinada à camada [Ui]. Esse tipo é o seguinte:

package dvp.rxswing.ui;

import dvp.rxswing.service.ServiceResponse;

import java.text.SimpleDateFormat;
import java.util.Calendar;

public class UiResponse {

  // ID do cliente
  private int idClient;
  // resposta do serviço
  private ServiceResponse serviceResponse;
  // nome do thread de observação
  private String observedOn;
  // hora da solicitação
  private String requestAt;
  // hora da resposta
  private String responseAt;

  // construtores

  public UiResponse() {
      // thread de observação
    observedOn = Thread.currentThread().getName();
    // hora da solicitação
    requestAt = getTimeStamp();
  }

  // métodos privados

  private String getTimeStamp() {
    return new SimpleDateFormat("hh:mm:ss:SSS").format(Calendar.getInstance().getTime());
  }

  // getters e setters
...
}

• os números aleatórios estão no campo da linha 13;
• os demais campos servem para especificar os threads de execução e de observação do observável do serviço assíncrono, bem como as horas da solicitação feita ao serviço e da resposta obtida;

A interface assíncrona é implementada pela seguinte classe [RxService]:

package dvp.rxswing.service;

import dvp.rxswing.ui.UiResponse;
import rx.Observable;

public class RxService implements IRxService {

  // serviço síncrono
  private IService service;

  // construtor
  public RxService(IService service) {
    this.service = service;
  }

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
      // cria-se um observável que emite o valor retornado pelo serviço síncrono
    return Observable.create(subscriber -> {
      try {
          // chamada síncrona
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        // passa-se o resultado para o observador
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
          // passa-se o erro para o observador
        subscriber.onError(e);
      } finally {
          // avisa-se ao observador que as emissões terminaram
        subscriber.onCompleted();
      }
    });
  }
}

• linhas 12-14: a classe [RxService] do serviço assíncrono é construída a partir de uma instância da interface síncrona [IService];
• linhas 20-33: construção do observável, resultado do método [getAleas];
• linha 22: o método síncrono [service.getAleas] é chamado. Seu resultado, do tipo [ServiceResponse], é incluído no objeto do tipo [UiResponse] a ser fornecido à camada [swing]. Esse objeto foi inicialmente passado nos parâmetros de chamada do método (último parâmetro, linha 17);
• linha 24: a resposta [UiResponse] é enviada ao observador (a camada [swing]). O objeto [UiResponse] não contém apenas as informações geradas pelo serviço síncrono da linha 22. Ele também contém outras informações geradas pelo método que chamou o método [getAleas] da linha 17. É por essa razão que esse método chamador passou o objeto [UiResponse] como parâmetro para o método [getAleas] (último parâmetro, linha 17);
• linha 30: não se esquece de sinalizar o fim das transmissões. Temos aqui um observável que emite apenas um valor: aquele retornado pelo serviço síncrono;
• linha 27: sinaliza-se ao observador um possível erro;

8.6. A interface gráfica

• a interface gráfica foi criada com o IDE [Netbeans], que possui um bom editor gráfico. Esse editor gerou o arquivo [AbstractJFrameAleas.form], que só pode ser utilizado por esse IDE;
• a classe [AbstractJFrameAleas] também foi gerada pelo editor gráfico do NetBeans. Em seguida, ela foi refatorada da seguinte maneira: os eventos da interface gráfica que queríamos gerenciar são tratados na classe [AbstractJFrameAleas] por meio de métodos abstratos implementados na classe filha [JFrameAleasEvents]. Por fim,
  • a classe abstrata [AbstractJFrameAleas] é responsável por construir e exibir a interface gráfica;
  • a classe filha [JFrameAleasEvents] é responsável por gerenciar os eventos dessa interface;

Os componentes da interface gráfica da aba [Request] são os seguintes:

n.º	tipo	nome	função
1	JTabbedPane	jTabbedPane1	um contêiner de guias. Contém duas guias (JPanel) [jPanelRequest] para a solicitação e [jPanelresponse] para a resposta;
2	JTextField	jTextFieldNbValeurs	o número de solicitações a serem feitas ao serviço de números aleatórios. No caso do serviço assíncrono executado no agendador [Schedulers.io], essas solicitações compartilharão um processador;
3	JTextField	jTextFieldA	limite a do intervalo [a,b]
4	JTextField	jTextFieldB	terminal b do intervalo [a,b]
5	JTextField	jTextFieldMinCount	terminal minCount do intervalo [minCount, maxCount]
6	JTextField	jTextFieldMaxCount	terminal maxCount do intervalo [minCount, maxCount]
7	JTextField	jTextFieldMinDelay	terminal minDelay do intervalo [minDelay, maxDelay]
8	JTextField	jTextFieldMaxDelay	terminal maxDelay do intervalo [minDelay, maxDelay]
9	JCheckBox	jCheckBoxRxSwing	se a caixa estiver marcada, as consultas são feitas na interface assíncrona. Caso contrário, são feitas na interface síncrona
10	JComboBox	jComboBoxSchedulers	no caso de solicitações assíncronas, elas serão executadas com o agendador selecionado aqui
11	JButton	jButtonGenerate	inicia a execução das consultas no serviço síncrono ou assíncrono

Os componentes da interface gráfica da aba [Response] são os seguintes:

n.º	tipo	nome	função
1	JLabel	jLabelDuree	o tempo total de execução, em milissegundos, das consultas
2	JLabel	jLabelNbReponses	o número total de respostas observadas (pode ser diferente do número de consultas, já que cada consulta pode fornecer vários valores a serem observados)
3	JList	jListNumbers	exibição dos valores observados (recebidos)
4	JButton	jButtonAnnuler	cancela as consultas em execução

8.7. Instanciação da interface gráfica

A classe [JFrameAleasEvents] gerencia os eventos da interface gráfica, especialmente o clique no botão [Générer]. Trata-se de uma classe executável que é iniciada no seguinte contexto:

public class JFrameAleasEvents extends AbstractJFrameAleas {

    private static final long serialVersionUID = 1L;
    // serviço de geração síncrona
    private IService service;
    // serviço de geração assíncrona
    private IRxService rxService;

    // os registros
    private int nbRequests;
    private int a;
    private int b;
    private int minDelay;
    private int maxDelay;
    private int minCount;
    private int maxCount;

    // mensagens de erro
    private final String jLabelNbValuesErrorText = "Tapez un nombre entier >=1";
    private final String jLabelCountErrorText = "minCount doit être >=0 et maxCount>=minCount ";
    private final String jLabelDelayErrorText = "minDelay doit être >=0 et maxDelay>=minDelay et  maxDelay<=5000";
    private final String jLabelIntervalErrorText = "a doit être >=0 et b>=a ";

    // assinaturas de observáveis
    protected List<Subscription> subscriptions = new ArrayList<Subscription>();
    // início e fim da execução
    private long debut;
    // mapeador jSON
    private ObjectMapper jsonMapper;
    // modelo de respostas
    private DefaultListModel<String> model;

    // construtor
    public JFrameAleasEvents() {
        // pai
        super();
        // local
        initJFrame();
        // serviços
        service = new Service();
        rxService = new RxService(service);
        // mapeador jSON
        jsonMapper = new ObjectMapper();
    }

    private void initJFrame() {
        // ocultamos as mensagens de erro
        jLabelCountError.setText("");
        jLabelDelayError.setText("");
        jLabelIntervalError.setText("");
        jLabelNbValuesError.setText("");
        // ocultar textos padrão
        jTextFieldA.setText("100");
        jTextFieldB.setText("200");
        jTextFieldMinCount.setText("5");
        jTextFieldMaxCount.setText("10");
        jTextFieldMinDelay.setText("100");
        jTextFieldMaxDelay.setText("500");
        jTextFieldNbValeurs.setText("10");
        jLabelDuree.setText("");
        // modelo de respostas
        model = new DefaultListModel<>();
        jListNumbers.setModel(model);
        // número de núcleos
        System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
    }

    public static void main(String args[]) {
        try {
            UIManager.setLookAndFeel(UIManager.getSystemLookAndFeelClassName());
        } catch (UnsupportedLookAndFeelException | ClassNotFoundException | InstantiationException
                | IllegalAccessException e) {
            System.out.println(e);
            System.exit(0);
        }

        /* Criar e exibir o formulário */
        java.awt.EventQueue.invokeLater(() -> {
            new JFrameAleasEvents().setVisible(true);
        });
    }

• linha 1: a classe [JFrameAleasEvents] estende a classe [AbstractJFrameAleas], que, por sua vez, estende a classe Swing [JFrame]. A classe [JFrameAleasEvents] é, portanto, uma janela Swing;
• linhas 68-75: o método [main] que será executado;
• linha 70: define o “look and feel” da interface gráfica;
• linha 79: o construtor da classe [JFrameAleasEvents] é chamado: a interface gráfica será construída e inicializada. Feito isso, ela é tornada visível;
• linhas 34-44: o construtor;
• linha 36: a chamada ao construtor pai inicializará a interface gráfica. Nesse momento, ela está exatamente como o desenvolvedor a projetou. Ela ainda não está visível;
• linha 38: alguns componentes da interface gráfica são inicializados;
• linha 40: instanciação do serviço síncrono;
• linha 41: instanciação do serviço assíncrono;

8.8. Execução de consultas síncronas

Ao clicar no botão [Générer], é executado o seguinte método [doGenerate]:

    @Override
    protected void doGenerate() {
        // dados válidos?
        if (!isPageValid()) {
            return;
        }
        // rx ou não?
        if (jCheckBoxRxSwing.isSelected()) {
            // solicitações assíncronas
            doGenerateWithRxService();
        } else {
            // solicitações síncronas
            doGenerateWithService();
        }
}

• linhas 4-6: verifica-se se as entradas do usuário são válidas. Não comentaremos o método [isPageValid]. Ele é básico;
• linha 8: verifica-se o estado da caixa de seleção RxSwing;
• linha 13: executam-se as consultas de forma síncrona;

O método [doGenerateWithService] é o seguinte:

    // geração síncrona
    private void doGenerateWithService() {
        // início da espera
        beginWaiting();
        try {
            for (int i = 0; i < nbRequests; i++) {
                // preparação da resposta
                UiResponse uiResponse = new UiResponse();
                // n.º do cliente
                uiResponse.setIdClient(i);
                // chamada síncrona
                uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
                // hora da resposta
                uiResponse.setResponseAt();
                // atualização do modelo do JList com as respostas recebidas
                model.add(0, jsonMapper.writeValueAsString(uiResponse));
                // atualização do número de respostas
                jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
            }
        } catch (JsonProcessingException | RuntimeException e) {
            JOptionPane.showMessageDialog(this, getInfoForThrowable("L'erreur suivante s'est produite", e), "Informations",
                    JOptionPane.PLAIN_MESSAGE);
        }
        // fim da espera
        endWaiting();
}

• linha 12: chamada síncrona ao serviço de geração de números aleatórios;
• a execução do método [doGenerateWithService] ocorre inteiramente na thread do loop de eventos do Swing. Enquanto o método não estiver concluído, a interface gráfica não processa nenhum novo evento. Ela fica congelada (frozen). Assim, por exemplo, as atualizações da interface gráfica nas linhas 16 e 18 nunca serão exibidas. Elas só ficarão visíveis com seus valores finais, e isso ocorrerá ao final da execução de todas as consultas;

O método [beginWaiting] (linha 4) é o seguinte:

    private void beginWaiting() {
        // botões
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // cursor de espera
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // zerar respostas
        model.clear();
        // assinaturas Rx
        subscriptions.clear();
        // exibindo a visualização das respostas
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // início da execução
        debut = new Date().getTime();
}

• linha 3: o botão [Générer] está oculto. Isso cria um evento que também só poderá ser executado ao final da execução de todas as consultas. Por isso, ele nunca aparece oculto, pois o método [endWaiting], na linha 25 do método [doGenerateWithService], o exibe novamente;
• linha 13: seleciona-se a guia [Response] para ver as respostas chegando. Mais uma vez, esse evento só será executado ao final da execução de todas as consultas, quando então veremos todas as respostas de uma só vez, embora quiséssemos vê-las chegando uma após a outra;

A interface síncrona apresenta claramente deficiências. Essas deficiências são superadas graças à interface assíncrona.

8.9. Execução de consultas assíncronas

O código para a execução das consultas assíncronas é o seguinte:

private void doGenerateWithRxService() {
        // início da espera
        beginWaiting();
        // vamos obter os números aleatórios na forma de um observável
        Observable<UiResponse> observable = Observable.empty();
        // Agendador de execução dos diferentes observáveis
        Scheduler[] schedulers = { Schedulers.io(), Schedulers.computation(), Schedulers.newThread(),
                Schedulers.trampoline(), Schedulers.immediate() };
        Scheduler scheduler = schedulers[jComboBoxSchedulers.getSelectedIndex()];
        // configuração dos observáveis
        for (int i = 0; i < nbRequests; i++) {
            // preparação da resposta
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // o observável está configurado para ser executado no agendador escolhido pelo usuário
            // em seguida, soma do observável obtido ao observável total
            observable = observable.mergeWith(
                    rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(scheduler));
        }
        // observador
        observable = observable.observeOn(SwingScheduler.getInstance());
        // por enquanto, apenas fizemos a configuração
        // ainda não foi feita nenhuma solicitação ao serviço síncrono de geração de números aleatórios
        // estamos nos inscrevendo no observável — é isso que vai provocar a chamada ao serviço síncrono de geração de números aleatórios
        try {
            // há apenas uma assinatura aqui — o resultado é uma subscrição
            subscriptions.add(observable.subscribe(
                    // notificação de emissão
                    uiResponse -> {
                        // a interface do usuário é atualizada com a resposta
                        // isso é possível porque a observação ocorre na thread da interface do usuário
                        updateUi(uiResponse);
                    } ,
                    // notificação de erro
                    th -> {
                        // caso de erro — exibe-se
                        String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
                        JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
                        // cancelamento de solicitações
                        doCancel();
                    } ,
                    // notificação [onCompleted]
                    // fim da espera
                    this::endWaiting));
        } catch (Throwable th) {
            // caso de exceção + geral — exibido
            String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
            JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
            // cancelamento das solicitações
            doCancel();
        }
    }

• linha 3: a interface gráfica é modificada para indicar que uma operação potencialmente demorada está em andamento;
• linha 5: cria-se um observável vazio. É esse observável que será monitorado pela camada [swing];
• linha 7: a tabela dos agendadores possíveis;
• linha 9: oferecemos ao usuário a possibilidade de escolher o agendador no qual as consultas serão executadas. Recuperamos o agendador escolhido por ele;
• linhas 11-19: cada uma das consultas retorna um observável cujos elementos são acumulados (mergeWith) (linha 17) no observável da linha 5;
• linhas 13-14: o objeto [UiResponse] é construído. Vale lembrar que esse objeto é tanto o parâmetro de entrada do método [RxService.getAleas] quanto seu resultado (linhas 17-18);
• linha 14: cada solicitação é identificada por seu número, aqui denominado [idClient]. Isso é necessário porque, em um ambiente assíncrono, a ordem de recebimento das respostas pode ser diferente da ordem de envio das solicitações. [idClient] permite saber a qual solicitação a resposta pertence;
• linhas 17-18: a solicitação assíncrona é feita como [rxService.getAleas]. Ela é executada no agendador escolhido pelo usuário. Seu resultado, do tipo Observable<UiResponse>, é acumulado com o observável da linha 5. É preciso ter em mente que o método [rxService.getAleas] é executado aqui e retorna um observável. Isso, no entanto, não significa que números aleatórios tenham sido obtidos. De fato, um observável só é executado quando alguém se inscreve nele. Esse ainda não é o caso;
• linha 21: esta é a instrução importante: solicita-se que a observação dos elementos emitidos pelo observável da linha 5 seja feita na thread da interface do usuário. Aqui, utiliza-se um agendador específico da biblioteca RxSwing;
• linhas 25-51: inscrevemos-nos no observável da linha 5. Só agora os números aleatórios serão solicitados ao serviço síncrono de geração desses números. O essencial está nas instruções das linhas 29-33. O restante lida essencialmente com os casos de erros e com a notificação [onCompleted] do observável;
• linhas 28-44: é preciso lembrar que solicitamos a observação do processo da linha 5 na thread da interface do usuário. Portanto, o código das linhas 28-44 é executado na thread da interface do usuário;
• linhas 29-33: processamos a notificação [onNext] do observável. Recebemos um tipo [UiResponse] emitido pelo processo observado. Esse é o resultado de uma das solicitações assíncronas. Atualizamos a interface gráfica com essa resposta;
• linhas 34-41: processa-se a notificação [onError] do observável. Exibe-se uma caixa de diálogo informando o erro (linhas 37-38) e, em seguida, cancelam-se as solicitações (linha 40);
• linhas 42-44: processa-se a notificação [onCompleted] do observável. Atualiza-se a interface gráfica para mostrar que o serviço solicitado foi concluído. A linha 44 também poderia ter sido escrita da seguinte forma

 ()->{endWaiting();}

Preferimos, neste caso, utilizar uma referência de método;

• linhas 45-51: algumas exceções não passam pelas linhas 34-41. Esse é o caso quando são enviadas muitas solicitações. Após ultrapassar um certo limite, que depende do ambiente de trabalho no momento da execução, ocorre um [StackOverflowError] que é interceptado pelas linhas 45-51;
• linha 27: a assinatura gera um tipo [Subscription], que é adicionado a uma lista de assinaturas. Essa lista terá, neste caso, apenas um elemento;

Na linha 32, atualiza-se a interface gráfica com o seguinte método [updateUi]:

    private void updateUi(UiResponse uiResponse) {
        // tempo de resposta
        uiResponse.setResponseAt();
        // thread de observação
        uiResponse.setObservedOn();
        // número de respostas
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // tempo de execução
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // adição da cadeia jSON da resposta ao modelo JList das respostas
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

Vemos aqui que os componentes da interface gráfica são atualizados (linhas 7, 9, 12). Para que isso seja possível, é obrigatório estar na thread da interface gráfica (event loop).

O método [endWaiting] é o seguinte:

    private void endWaiting() {
        // botão [Générer] visível
        jButtonGenerate.setVisible(true);
        // botão [Annuler] oculto
        jButtonCancel.setVisible(false);
        // cursor de espera oculto
        jTabbedPane1.setCursor(Cursor.getDefaultCursor());
        // aba de respostas selecionada
        jTabbedPane1.setSelectedIndex(1);
        // tempo desde a última atualização
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
}

O método [doCancel] é chamado quando ocorre um erro na execução das solicitações assíncronas ou quando o usuário clica no botão [Annuler]. Seu código é o seguinte:

// assinaturas dos observáveis
    private List<Subscription> subscriptions = new ArrayList<Subscription>();
....

    @Override
    protected void doCancel() {
        // fim da espera
        endWaiting();
        // no caso de assinaturas
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
            //subscriptions.forEach(s -> s.unsubscribe());
        }
    }

• linha 2: [subscriptions] é uma lista de uma assinatura;
• linha 11: todas as assinaturas são canceladas;
• linha 12: outra versão da linha 11. O método [forEach] espera aqui uma instância do tipo Consumer<Subscription> (ver parágrafo 4.4);

Voltemos ao código do método [doGenerateWithService]: ele pode ser dividido em duas etapas:

1. etapa de configuração dos observáveis. Isso é feito na thread do chamador do método [doGenerateWithService], ou seja, a thread da interface do usuário;
2. a assinatura que provocará a execução dos observáveis;

Se os observáveis tiverem como agendador um dos agendadores [Schedulers.computation(), Scheduler.io(), Schedulers.newThread()], eles serão executados fora da thread da interface do usuário. Essas diferentes threads disputarão o(s) núcleo(s) da máquina. Como as consultas são operações demoradas (várias centenas de milissegundos), o método [doGenerateWithService] executado na thread da interface do usuário será concluído antes que as consultas tenham retornado suas respostas. No entanto, esse método havia sido executado no evento de clique no botão [Générer]. Com esse evento processado, a thread da interface do usuário poderá passar para o processamento dos eventos seguintes. Existem vários deles. Assim, o método [beginWaiting] havia posicionado vários:

    private void beginWaiting() {
        // botões
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // cursor de espera
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // apagar respostas
        model.clear();
        // assinaturas Rx
        subscriptions.clear();
        // exibe a visualização das respostas
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // início da execução
        debut = new Date().getTime();
}

Praticamente todas as linhas desse código têm um efeito na interface gráfica. Essa atualização não ocorre imediatamente: os eventos são colocados na fila do loop de eventos. Assim que o evento de clique no botão [Générer] for processado, esses eventos são executados por sua vez e o usuário pode ver a interface gráfica mudar:

• a aba [Response] é exibida (linha 13) e um cursor de espera é associado a ela (linha 6)
• seu botão [Annuler] é exibido (linha 4) e o usuário poderá clicar nele;
• o JList das respostas é esvaziado (linha 9);
• o JLabel do número de respostas exibe 0;
• o JLabel da duração de execução exibe uma string vazia;

Durante todo o tempo de execução das consultas, a thread do UI tem acesso regular ao processador. Assim, ela pode processar os eventos em espera. Entre eles, estão aqueles definidos pelo método [updateUi]:

    private void updateUi(UiResponse uiResponse) {
        // hora da resposta
        uiResponse.setResponseAt();
        // thread de observação
        uiResponse.setObservedOn();
        // número de respostas
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // duração da execução
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // adição da cadeia jSON da resposta ao modelo JList das respostas
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

Quando a thread da interface do usuário está em execução:

• o JLabel, que representa o número de respostas, é atualizado (linha 7);
• o JLabel do tempo de execução é atualizado (linha 9);
• o JList das respostas é atualizado por meio de seu modelo (linha 12);

Assim, o usuário vê o andamento da execução das consultas. Além disso, ele pode cancelá-las por meio do botão [Annuler]. É aí que reside todo o interesse de ter serviços assíncronos na camada [swing], e o RxJava é a tecnologia ideal para implementá-los.

Para concluir, observemos que, se o usuário escolher um dos agendadores [Schedulers.immediate(), Schedulers.trampoline()], os observáveis serão executados na mesma thread que o chamador, ou seja, a thread da interface do usuário. Assim, voltamos a um funcionamento síncrono.

Os resultados obtidos com os diferentes agendadores foram apresentados nos parágrafos 2.8.1, 2.8.2, 2.8.3 e 2.8.4.