8. RxJava no ambiente Swing

8.1. Introduction

Vamos agora voltar à aplicação Swing apresentada no parágrafo 2.

Para trabalhar com o RxJava num ambiente Swing, utilizaremos a biblioteca RxSwing, que adiciona ao RxJava classes e interfaces úteis num ambiente Swing. Para tal, o ficheiro 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;

Iremos utilizar apenas um único objeto específico de RxSwing: o agendador [SwingScheduler.getInstance()], que executa/observa os observáveis no thread do evento loop do Swing. Vamos utilizá-lo exclusivamente para observar observáveis que se executam noutras threads que não a do evento loop. Recorde-se a arquitetura da aplicação de exemplo:

• a camada de serviço assíncrona apresenta métodos que devolvem observáveis. Executamos esses observáveis em threads diferentes da do evento loop. Assim, a interface gráfica não fica bloqueada. Pode reagir às solicitações do utilizador. A forma mais óbvia é permitir que o utilizador clique num botão [Annuler] para interromper uma operação assíncrona demasiado demorada. Para que isso seja possível, basta que a interface gráfica esteja congelada (frozen);
• a camada Swing pretende explorar os resultados devolvidos pelas operações assíncronas e, a partir deles, atualizar a interface gráfica. No entanto, isto só pode ser feito no thread do event loop. Para tal, esses resultados são observados no agendador [SwingScheduler.getInstance()];

Assim, no código de gestão 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, consoante os casos de utilização.

Convida-se o leitor a reler o parágrafo 2. Dispõe agora dos conhecimentos necessários para o compreender na íntegra.

8.2. A estrutura do código

O código implementa a seguinte arquitetura:

O projeto IntelliJ IDEA que implementa esta 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íncrono 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 n próprio um número aleatório no intervalo [minCount, maxCount]
  // os números são gerados após um atraso 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]
    // são gerados n números, sendo n próprio um número aleatório no intervalo [minCount, maxCount]
    // os números são gerados após um atraso 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: estende a classe [RuntimeException]. Trata-se, portanto, de uma exceção não controlada;
• linha 7: acrescenta um código de erro à sua classe pai (0 = sem erro);

8.5. O serviço assíncrono

A camada de serviço assíncrono 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]
  // são gerados n números, sendo n próprio um número aleatório no intervalo [minCount, maxCount]
  // os números são gerados após um atraso 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: o método [getAleas] do serviço devolve agora um observável;

O método [getAleas] devolve uma resposta do tipo [UiResponse] destinada à camada [Ui]. Este 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 encontram-se no campo da linha 13;
• os restantes 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 devolvido 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 ao observador
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
          // passa-se o erro ao observador
        subscriber.onError(e);
      } finally {
          // avisa-se o observador de 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. O seu resultado, do tipo [ServiceResponse], é incluído no objeto do tipo [UiResponse] a ser fornecido à camada [swing]. Este 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 criadas pelo serviço síncrono da linha 22. Contém também outras informações criadas pelo método que chamou o método [getAleas] da linha 17. É por esta razão que este método chamador passou o objeto [UiResponse] como parâmetro ao método [getAleas] (último parâmetro, linha 17);
• linha 30: não nos esquecemos de sinalizar o fim das emissões. Temos aqui um observável que emite apenas um valor: aquele devolvido pelo serviço síncrono;
• linha 27: indica-se ao observador um eventual erro;

8.6. A interface gráfica

• A interface gráfica foi criada com o IDE [Netbeans], que dispõe de um bom editor gráfico. Este editor gerou o ficheiro [AbstractJFrameAleas.form], que só pode ser utilizado por este IDE;
• a classe [AbstractJFrameAleas] foi igualmente gerada pelo editor gráfico do NetBeans. Posteriormente, foi refatorada da seguinte forma: os eventos da interface gráfica que se pretendia gerir são tratados na classe [AbstractJFrameAleas] por métodos abstratos implementados na classe filha [JFrameAleasEvents]. Por fim,
  • A classe abstrata [AbstractJFrameAleas] encarrega-se de construir e apresentar a interface gráfica;
  • a classe filha [JFrameAleasEvents] é responsável pela gestão dos eventos desta última;

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

n.º	tipo	nome	função
1	JTabbedPane	jTabbedPane1	um contentor de separadores. Contém dois separadores (JPanel) [jPanelRequest] para o pedido, [jPanelresponse] para a resposta;
2	JTextField	jTextFieldNbValeurs	o número de pedidos a efetuar ao serviço de números aleatórios. No caso do serviço assíncrono executado no agendador [Schedulers.io], estes pedidos irão partilhar 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, os pedidos são enviados para a interface assíncrona. Caso contrário, são enviados para a interface síncrona
10	JComboBox	jComboBoxSchedulers	No caso de pedidos assíncronos, estes serão executados com o agendador aqui selecionado
11	JButton	jButtonGenerate	inicia a execução das consultas no serviço síncrono ou assíncrono

Os componentes da interface gráfica do separador [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 pedidos, uma vez que cada pedido pode fornecer vários valores a observar)
3	JList	jListNumbers	exibição dos valores observados (recebidos)
4	JButton	jButtonAnnuler	Anula as consultas em execução

8.7. Instanciação da interface gráfica

A classe [JFrameAleasEvents] gere os eventos da interface gráfica e, nomeadamente, 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 registos
    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 ";

    // subscrições 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() {
        // ocultar mensagens de erro
        jLabelCountError.setText("");
        jLabelDelayError.setText("");
        jLabelIntervalError.setText("");
        jLabelNbValuesError.setText("");
        // ocultar os textos por predefiniçã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 apresentar 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 vai 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 vai ser construída e inicializada. Feito isto, é tornada visível;
• linhas 34-44: o construtor;
• linha 36: a chamada ao construtor pai irá inicializar a interface gráfica. Nesse momento, esta apresenta-se tal como foi concebida pelo programador. 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 utilizador são válidas. Não iremos comentar o método [isPageValid]. Trata-se de um método 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 evento loop do Swing. Enquanto o método não estiver concluído, a interface gráfica não processa nenhum novo evento. Fica congelada (frozen). Assim, por exemplo, as atualizações da interface gráfica nas linhas 16 e 18 nunca serão visíveis. Só serão visíveis com os seus valores finais, no 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());
        // limpar respostas
        model.clear();
        // assinaturas Rx
        subscriptions.clear();
        // exibir a vista 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. Isto cria um evento que também só poderá ser executado no final da execução de todas as consultas. Por isso, nunca o vemos oculto, pois o método [endWaiting], na linha 25 do método [doGenerateWithService], volta a exibi-lo;
• linha 13: seleciona-se o separador [Response] para ver as respostas a chegar. Mais uma vez, este evento só será executado no final da execução de todas as consultas, altura em que se verão todas as respostas de uma só vez, quando o que se pretendia era vê-las a chegar uma a seguir à outra;

A interface síncrona apresenta claramente algumas limitações. Estas são superadas graças à interface assíncrona.

8.9. Execução de pedidos assíncronos

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 sob a 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 utilizador
            // em seguida, acumulação do observável obtido ao observável global
            observable = observable.mergeWith(
                    rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(scheduler));
        }
        // observador
        observable = observable.observeOn(SwingScheduler.getInstance());
        // por enquanto, limitámo-nos apenas à configuração
        // ainda não foi feita nenhuma solicitação ao serviço síncrono de geração de números aleatórios
        // subscrevemos o observável — é isso que vai provocar a chamada ao serviço síncrono de geração de números aleatórios
        try {
            // aqui existe apenas uma subscrição — o resultado é uma subscrição
            subscriptions.add(observable.subscribe(
                    // notificação de emissão
                    uiResponse -> {
                        // atualiza-se a interface do utilizador com a resposta
                        // isto é possível porque a observação ocorre no thread da interface do utilizador
                        updateUi(uiResponse);
                    } ,
                    // notificação de erro
                    th -> {
                        // caso de erro — é apresentado
                        String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
                        JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
                        // cancelamento de pedidos
                        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);
            // cancelam-se as solicitações
            doCancel();
        }
    }

• linha 3: a interface gráfica é alterada para indicar que está em curso uma operação potencialmente demorada;
• linha 5: cria-se um observável vazio. É este observável que será observado pela camada [swing];
• linha 7: a tabela dos agendadores possíveis;
• linha 9: demos ao utilizador a possibilidade de escolher o agendador no qual executar as consultas. Recuperamos o agendador da sua escolha;
• linhas 11-19: cada uma das consultas devolve 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. Recorde-se que este objeto é simultaneamente o parâmetro de entrada do método [RxService.getAleas] e o seu resultado (linhas 17-18);
• linha 14: cada pedido é identificado pelo seu número, aqui designado por [idClient]. Isto é necessário porque, num ambiente assíncrono, a ordem de receção das respostas pode ser diferente da ordem de envio dos pedidos. O [idClient] permite saber a que pedido pertence a resposta;
• linhas 17-18: a solicitação assíncrona é efetuada como [rxService.getAleas]. É executada no agendador escolhido pelo utilizador. O seu resultado, do tipo Observable<UiResponse>, é acumulado com o observável da linha 5. É importante ter em conta que o método [rxService.getAleas] é aqui executado e devolve um observável. No entanto, isso não significa que tenham sido obtidos números aleatórios. Com efeito, um observável só é executado quando se subscreve o mesmo. Ainda não é esse 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 no thread da interface do utilizador. Aqui, utiliza-se um agendador específico da biblioteca RxSwing;
• linhas 25-51: subscreve-se o observável da linha 5. Só agora é que 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 trata essencialmente dos casos de erros e da notificação [onCompleted] do observável;
• linhas 28-44: é importante lembrar que solicitámos a observação do processo da linha 5 no thread da interface do utilizador. Assim, o código das linhas 28-44 é executado no thread da interface do utilizador;
• linhas 29-33: processamos a notificação [onNext] do observável. Recebemos um tipo [UiResponse] emitido pelo processo observado. Trata-se do resultado de uma das requisições assíncronas. Atualizamos a interface gráfica com esta resposta;
• linhas 34-41: processa-se a notificação [onError] do observável. Exibe-se uma caixa de diálogo com 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 indicar que o serviço solicitado foi concluído. A linha 44 também poderia ter sido escrita da seguinte forma

 ()->{endWaiting();}

Preferiu-se aqui utilizar uma referência de método;

• linhas 45-51: algumas exceções não passam pelas linhas 34-41. É o que acontece quando se enviam demasiadas solicitações. Após ultrapassar um determinado limite, que depende do ambiente de trabalho no momento da execução, surge um [StackOverflowError] que é interceptado pelas linhas 45-51;
• linha 27: a subscrição gera um tipo [Subscription] que é adicionado a uma lista de subscrições. Esta 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) {
        // hora da 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 no thread da UI (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());
        // separador «Respostas» selecionado
        jTabbedPane1.setSelectedIndex(1);
        // data da última atualização
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
}

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

// as subscrições dos observáveis
    private List<Subscription> subscriptions = new ArrayList<Subscription>();
....

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

• linha 2: [subscriptions] é uma lista de uma subscrição;
• linha 11: todas as subscrições 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]: este pode ser dividido em duas etapas:

1. etapa de configuração dos observáveis. Isto é feito no thread do chamador do método [doGenerateWithService], ou seja, o thread da interface do utilizador;
2. a subscrição que irá provocar a execução dos observáveis;

Se os observáveis tiverem como agendador um dos agendadores [Schedulers.computation(), Scheduler.io(), Schedulers.newThread()], então serão executados fora do thread da interface do utilizador. Estes diferentes threads irão disputar 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 utilizador terminará antes de as consultas terem devolvido as suas respostas. Ora, este método tinha sido executado no evento de clique no botão [Générer]. Uma vez processado este evento, a thread da IU poderá passar ao processamento dos eventos seguintes. Existem vários. Assim, o método [beginWaiting] tinha-os posicionado:

    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());
        // limpar respostas
        model.clear();
        // subscrições Rx
        subscriptions.clear();
        // exibir a vista das respostas
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // início da execução
        debut = new Date().getTime();
}

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

• o separador [Response] é apresentado (linha 13) e é-lhe associado um cursor de espera (linha 6)
• o seu botão [Annuler] é apresentado (linha 4) e o utilizador poderá clicar nele;
• o JList das respostas é esvaziado (linha 9);
• o JLabel do número de respostas apresenta o valor 0;
• o JLabel da duração de execução apresenta uma cadeia vazia;

Durante todo o tempo de execução das consultas, o thread do UI tem acesso regular ao processador. Pode, assim, processar os eventos em espera. Entre estes, encontram-se os definidos pelo método [updateUi]:

    private void updateUi(UiResponse uiResponse) {
        // hora da resposta
        uiResponse.setResponseAt();
        // tópico 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 utilizador está ativa:

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

Desta forma, o utilizador vê o progresso da execução das consultas. Além disso, pode cancelá-las através do botão [Annuler]. É aí que reside todo o interesse de ter serviços assíncronos na frente da camada [swing], sendo que o RxJava é uma tecnologia de eleição para implementar esses serviços.

Para concluir, note-se que, se o utilizador escolher um dos agendadores [Schedulers.immediate(), Schedulers.trampoline()], os observáveis são então executados no mesmo thread que o chamador, ou seja, o thread da interface do utilizador. Passa-se, assim, 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.