8. RxJava no ambiente Swing

8.1. Introdução

Aqui, vamos revisitar a aplicação Swing apresentada na Secção 2.

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

Iremos utilizar apenas um único objeto específico do RxSwing: o agendador [SwingScheduler.getInstance()], que executa/observa observáveis na thread do ciclo de eventos do Swing. Iremos utilizá-lo exclusivamente para observar observáveis em execução em threads que não sejam a do ciclo de eventos. Vamos rever a arquitetura da aplicação de exemplo:

• a camada de serviço assíncrono fornece métodos que devolvem observáveis. Executamos estes observáveis em threads diferentes da thread do ciclo de eventos. Desta forma, a GUI permanece responsiva. Pode reagir à entrada do utilizador. O exemplo mais óbvio é permitir que o utilizador clique num botão [Cancelar] para interromper uma operação assíncrona que está a demorar demasiado tempo. Para que isto funcione, a GUI deve estar congelada;
• a camada Swing precisa de processar os resultados devolvidos pelas operações assíncronas e utilizá-los para atualizar a GUI. No entanto, isto só pode ser feito na thread do ciclo de eventos. Para o conseguir, estes resultados são observados no agendador [SwingScheduler.getInstance()];

Assim, no código de tratamento de eventos da GUI, a interação com a camada assíncrona [rxService] assume a seguinte forma:

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

onde o agendador [Schedulers.computation()] pode ser substituído por outro agendador, dependendo do caso de uso.

Convidamos o leitor a reler o parágrafo 2. Agora já possui os 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. Executar o projeto

Para executar o projeto no IntelliJ IDEA, siga estes passos:

8.4. O serviço síncrono

A camada de serviços síncronos fornece a seguinte interface [IService]:

package dvp.rxswing.service;
 
public interface IService {
  // random numbers in the [a,b] interval
  // n numbers are generated with n itself a random number in the interval [minCount, maxCount]
  // numbers are generated after a delay of milliseconds,
  // where [delay] is itself a random number in the interval [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 {
 
  // service waiting time
  private int delay;
  // random numbers
  private List<Integer> aleas;
  // execution thread
  private String executedOn;
 
  // manufacturers
 
  public ServiceResponse() {
      // execution thread
    executedOn = Thread.currentThread().getName();
  }
 
  public ServiceResponse(int delay, List<Integer> aleas) {
      // local builder
    this();
    // other initializations
    this.delay = delay;
    this.aleas = aleas;
  }
 
  // getters and 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) {
    // random numbers in the [a,b] interval
    // n numbers are generated with n itself a random number in the interval [minCount, maxCount]
    // numbers are generated after a delay of milliseconds,
    // where [delay] is itself a random number in the interval [minDelay, maxDelay]
 
    // some checks
    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;
    }
    // mistakes?
    if (!messages.isEmpty()) {
      throw new AleasException(String.join(" [---] ", messages), erreur);
    }
    // random number generator
    Random random = new Random();
    // waiting?
    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);
      }
    }
    // result generation
    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));
    }
    // return result
    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;
    // error code
  private int code;
 
  // manufacturers
  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 and setters
...
}

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

8.5. O serviço assíncrono

A camada de serviços assíncronos fornece a seguinte interface [IRxService]:

package dvp.rxswing.service;
 
import dvp.rxswing.ui.UiResponse;
import rx.Observable;
 
public interface IRxService {
  // random numbers in the [a,b] interval
  // n numbers are generated with n itself a random number in the interval [minCount, maxCount]
  // numbers are generated after a delay of milliseconds,
  // where [delay] is itself a random number in the interval [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 agora retorna um observável;

O método [getAleas] retorna 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 {
 
  // customer id
  private int idClient;
  // service response
  private ServiceResponse serviceResponse;
  // observation thread name
  private String observedOn;
  // query time
  private String requestAt;
  // response time
  private String responseAt;
 
  // manufacturers
 
  public UiResponse() {
      // observation thread
    observedOn = Thread.currentThread().getName();
    // query time
    requestAt = getTimeStamp();
  }
 
  // private methods
 
  private String getTimeStamp() {
    return new SimpleDateFormat("hh:mm:ss:SSS").format(Calendar.getInstance().getTime());
  }
 
  // getters and setters
...
}

• Os números aleatórios estão no campo na linha 13;
• os outros campos servem para especificar os threads de execução e observação do observável do serviço assíncrono, bem como os carimbos de data/hora do pedido feito ao serviço e da resposta recebida;

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 {
 
  // synchronous service
  private IService service;
 
  // manufacturer
  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) {
      // we create an observable emitting the value rendered by the synchronous service
    return Observable.create(subscriber -> {
      try {
          // synchronous call
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        // the result is passed on to the observer
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
          // we pass the error to the observer
        subscriber.onError(e);
      } finally {
          // the observer is informed that the emissions are finished
        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, o 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: o [UiResponse] é enviado ao observador (a camada [swing]). O objeto [UiResponse] contém não só a informação gerada pelo serviço síncrono na linha 22, mas também outra informação gerada pelo método que chama o método [getAleas] na linha 17. É por isso que o método de chamada passou o objeto [UiResponse] como parâmetro para o método [getAleas] (último parâmetro, linha 17);
• linha 30: não nos esquecemos de sinalizar o fim das emissões. Aqui temos um observável que emite apenas um valor: aquele devolvido pelo serviço síncrono;
• linha 27: notificamos o observador de quaisquer erros;

8.6. A interface gráfica do utilizador

• A interface gráfica do utilizador foi criada utilizando o IDE [NetBeans], que possui um bom editor gráfico. Este editor gerou o ficheiro [AbstractJFrameAleas.form], que só pode ser utilizado por este IDE;
• A classe [AbstractJFrameAleas] também foi gerada pelo editor gráfico do NetBeans. Foi então refatorada da seguinte forma: os eventos da GUI que pretendíamos tratar são processados na classe [AbstractJFrameAleas] através de métodos abstratos implementados na classe filha [JFrameAleasEvents]. Por fim,
  • a classe abstrata [AbstractJFrameAleas] é responsável pela construção e exibição da interface gráfica;
  • a classe filha [JFrameAleasEvents] lida com os seus eventos;

Os componentes da GUI do separador [Request] são os seguintes:

N.º	tipo	nome	função
1	JTabbedPane	jTabbedPane1	Um contentor com separadores. Contém dois separadores (JPanel): [jPanelRequest] para o pedido e [jPanelResponse] para a resposta;
2	JTextField	jTextFieldNbValues	o número de pedidos a serem feitos ao serviço de números aleatórios. No caso do serviço assíncrono a ser executado no agendador [Schedulers.io], estes pedidos irão partilhar um processador;
3	JTextField	jTextFieldA	ponto final a do intervalo [a,b]
4	JTextField	jTextFieldB	ponto final b do intervalo [a,b]
5	JTextField	jTextFieldMinCount	minCount extremidade do intervalo [minCount, maxCount]
6	JTextField	jTextFieldMaxCount	limite superior do intervalo [minCount, maxCount]
7	JTextField	jTextFieldMinDelay	limite minDelay do intervalo [minDelay, maxDelay]
8	JTextField	jTextFieldMaxDelay	limite maxDelay do intervalo [minDelay, maxDelay]
9	JCheckBox	jCheckBoxRxSwing	Se a caixa de seleção estiver marcada, as solicitações são feitas à interface assíncrona. Caso contrário, são feitas à interface síncrona
10	JComboBox	jComboBoxSchedulers	Para pedidos assíncronos, estes serão executados utilizando o agendador selecionado aqui
11	JButton	jButtonGenerate	inicia a execução de solicitações ao serviço síncrono ou assíncrono

Os componentes da interface gráfica da guia [Resposta] são os seguintes:

N.º	tipo	nome	função
1	JLabel	jLabelDuration	o tempo total de execução, em milissegundos, das solicitações
2	JLabel	jLabelNbResponses	o número total de respostas observadas (pode diferir do número de pedidos, uma vez que cada pedido pode devolver vários valores a serem observados)
3	JList	jListNumbers	exibição dos valores observados (recebidos)
4	JButton	jButtonCancel	cancela os pedidos atualmente em execução

8.7. Instanciação da interface gráfica do utilizador

A classe [JFrameAleasEvents] trata dos eventos da GUI, incluindo cliques no botão [Generate]. É uma classe executável que é executada no seguinte contexto:

public class JFrameAleasEvents extends AbstractJFrameAleas {
 
    private static final long serialVersionUID = 1L;
    // synchronous generation service
    private IService service;
    // asynchronous generation service
    private IRxService rxService;
 
    // seizures
    private int nbRequests;
    private int a;
    private int b;
    private int minDelay;
    private int maxDelay;
    private int minCount;
    private int maxCount;
 
    // error messages
    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 ";
 
    // subscriptions to observables
    protected List<Subscription> subscriptions = new ArrayList<Subscription>();
    // start-end of execution
    private long debut;
    // mapper jSON
    private ObjectMapper jsonMapper;
    // answer model
    private DefaultListModel<String> model;
 
    // manufacturer
    public JFrameAleasEvents() {
        // parent
        super();
        // local
        initJFrame();
        // services
        service = new Service();
        rxService = new RxService(service);
        // mapper jSON
        jsonMapper = new ObjectMapper();
    }
 
    private void initJFrame() {
        // hide error messages
        jLabelCountError.setText("");
        jLabelDelayError.setText("");
        jLabelIntervalError.setText("");
        jLabelNbValuesError.setText("");
        // hide texts by default
        jTextFieldA.setText("100");
        jTextFieldB.setText("200");
        jTextFieldMinCount.setText("5");
        jTextFieldMaxCount.setText("10");
        jTextFieldMinDelay.setText("100");
        jTextFieldMaxDelay.setText("500");
        jTextFieldNbValeurs.setText("10");
        jLabelDuree.setText("");
        // answer model
        model = new DefaultListModel<>();
        jListNumbers.setModel(model);
        // number of hearts
        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);
        }
 
        /* Create and display the form */
        java.awt.EventQueue.invokeLater(() -> {
            new JFrameAleasEvents().setVisible(true);
        });
    }

• linha 1: a classe [JFrameAleasEvents] estende a classe [AbstractJFrameAleas], que por sua vez estende a classe [JFrame] do Swing. A classe [JFrameAleasEvents] é, portanto, uma janela do Swing;
• linhas 68–75: o método [main] que será executado;
• linha 70: define a aparência da GUI;
• linha 79: o construtor da classe [JFrameAleasEvents] é chamado: a GUI será construída e inicializada. Assim que isso estiver feito, ela é tornada visível;
• linhas 34–44: o construtor;
• linha 36: a chamada ao construtor pai irá inicializar a GUI. Nesta altura, tem exatamente o aspeto que o programador concebeu. Ainda não está visível;
• linha 38: determinados componentes da GUI 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 pedidos síncronos

Clicar no botão [Gerar] aciona a execução do seguinte método [doGenerate]:

    @Override
    protected void doGenerate() {
        // saisies valides ?
        if (!isPageValid()) {
            return;
        }
        // rx ou pas ?
        if (jCheckBoxRxSwing.isSelected()) {
            // requêtes asynchrones
            doGenerateWithRxService();
        } else {
            // requêtes synchrones
            doGenerateWithService();
        }
}

• linhas 4–6: verificamos se a entrada do utilizador é válida. Não vamos comentar o método [isPageValid]. É básico;
• linha 8: verificamos o estado da caixa de seleção RxSwing;
• linha 13: executamos os pedidos de forma síncrona;

O método [doGenerateWithService] é o seguinte:

    // synchronous generation
    private void doGenerateWithService() {
        // start waiting
        beginWaiting();
        try {
            for (int i = 0; i < nbRequests; i++) {
                // response preparation
                UiResponse uiResponse = new UiResponse();
                // customer no
                uiResponse.setIdClient(i);
                // synchronous call
                uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
                // response time
                uiResponse.setResponseAt();
                // update the JList model with the responses received
                model.add(0, jsonMapper.writeValueAsString(uiResponse));
                // update number of responses
                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);
        }
        // end waiting
        endWaiting();
}

• linha 12: chamada síncrona ao serviço de geração de números aleatórios;
• o método [doGenerateWithService] é executado inteiramente dentro da thread do ciclo de eventos do Swing. Até que o método termine, a GUI não processa quaisquer novos eventos. Fica congelada. Assim, por exemplo, as atualizações da GUI nas linhas 16 e 18 nunca serão visíveis. Só serão visíveis com os seus valores finais, e isto ocorrerá no final da execução de todos os pedidos;

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

    private void beginWaiting() {
        // buttons
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // wait slider
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // raz answers
        model.clear();
        // rx subscriptions
        subscriptions.clear();
        // the response view is displayed
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // start of execution
        debut = new Date().getTime();
}

• linha 3: o botão [Generate] está oculto. Isto cria um evento que também só pode ser executado depois de todas as solicitações terem terminado de ser processadas. No entanto, nunca o vemos oculto, porque o método [endWaiting] na linha 25 do método [doGenerateWithService] o exibe novamente;
• linha 13: selecionamos o separador [Response] para ver as respostas a chegar. Mais uma vez, este evento só será executado depois de todas as solicitações terem terminado, altura em que veremos todas as respostas de uma só vez, quando o que queríamos era vê-las a chegar uma a seguir à outra;

A interface síncrona tem claramente deficiências. Estas são superadas pela interface assíncrona.

8.9. Executar Pedidos Assíncronos

O código para executar pedidos assíncronos é o seguinte:

private void doGenerateWithRxService() {
        // début attente
        beginWaiting();
        // on va obtenir les nombres aléatoires sous la forme d'un observable
        Observable<UiResponse> observable = Observable.empty();
        // Schéduler d'exécution des différents observables
        Scheduler[] schedulers = { Schedulers.io(), Schedulers.computation(), Schedulers.newThread(),
                Schedulers.trampoline(), Schedulers.immediate() };
        Scheduler scheduler = schedulers[jComboBoxSchedulers.getSelectedIndex()];
        // configuration des observables
        for (int i = 0; i < nbRequests; i++) {
            // préparation réponse
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // l'observable est configuré pour s'exécuter sur le schéduler choisi par l'utilisateur
            // puis cumul de l'observable obtenu à l'observable du tout
            observable = observable.mergeWith(
                    rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(scheduler));
        }
        // observateur
        observable = observable.observeOn(SwingScheduler.getInstance());
        // pour l'instant, on a juste fait de la configuration
        // aucune requête n'a encore été faite au service synchrone de génération des nombres aléatoires
        // on s'abonne à l'observable - c'est ce qui va provoquer l'appel au service synchrone de génération des nombres aléatoires
        try {
            // il n'y a ici qu'un abonnement - le résultat est une souscription
            subscriptions.add(observable.subscribe(
                    // notification d'émission
                    uiResponse -> {
                        // on met à jour l'Ui avec la réponse
                        // ceci est possible car l'observation a lieu dans le thread de l'Ui
                        updateUi(uiResponse);
                    } ,
                    // notification d'erreur
                    th -> {
                        // cas d'erreur - on l'affiche
                        String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
                        JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
                        // annulation requêtes
                        doCancel();
                    } ,
                    // notification [onCompleted]
                    // fin de l'attente
                    this::endWaiting));
        } catch (Throwable th) {
            // cas d'exception + générale - on l'affiche
            String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
            JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
            // on annule les requêtes
            doCancel();
        }
    }

• linha 3: a GUI é atualizada para indicar que está em curso uma operação que poderá demorar algum tempo;
• linha 5: é criado um observável vazio. Este observável será observado pela camada [Swing];
• linha 7: o conjunto de possíveis agendadores;
• 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 pedido devolve um observável cujos elementos são fundidos (mergeWith) (linha 17) no observável da linha 5;
• linhas 13–14: o objeto [UiResponse] é construído. Recorde-se que este objeto é tanto o parâmetro de entrada do método [RxService.getAleas] como o seu resultado (linhas 17–18);
• Linha 14: Cada pedido é identificado por um número, aqui referido como [idClient]. Isto é necessário porque, num ambiente assíncrono, a ordem em que as respostas são recebidas pode diferir da ordem em que os pedidos foram enviados. [idClient] permite-nos determinar a que pedido a resposta pertence;
• Linhas 17–18: A solicitação assíncrona é feita [rxService.getRandom]. Ela é executada no agendador escolhido pelo utilizador. O seu resultado, do tipo Observable<UiResponse>, é combinado com o observável da linha 5. É importante notar que o método [rxService.getAleas] é executado aqui e retorna um observável. No entanto, isto não significa que tenham sido gerados números aleatórios. Na verdade, um observável só é executado quando subscrito. Este ainda não é o caso;
• linha 21: esta é a instrução chave: especificamos que os elementos emitidos pelo observável da linha 5 devem ser observados na thread da interface do utilizador. Aqui, usamos um agendador específico da biblioteca RxSwing;
• linhas 25–51: subscrevemos o observável da linha 5. Só agora é que os números aleatórios serão solicitados ao serviço síncrono que os gera. A parte fundamental está nas instruções das linhas 29–33. O resto trata principalmente de casos de erro e da notificação [onCompleted] do observável;
• linhas 28–44: Lembre-se de que solicitámos observar o processo da linha 5 na thread da interface do utilizador. Portanto, o código nas linhas 28–44 é executado na thread da interface do utilizador;
• linhas 29–33: Tratamos a notificação [onNext] do observável. Recebemos um tipo [UiResponse] emitido pelo processo observado. Este é o resultado de uma das solicitações assíncronas. Atualizamos a interface do utilizador com esta resposta;
• linhas 34–41: tratamos da notificação [onError] do observável. Exibimos uma caixa de diálogo mostrando o erro (linhas 37–38) e, em seguida, cancelamos as solicitações (linha 40);
• linhas 42–44: tratamos da notificação [onCompleted] do observável. Atualizamos a GUI para mostrar que o serviço solicitado foi concluído. A linha 44 também poderia ter sido escrita da seguinte forma

 ()->{endWaiting();}

Aqui, optámos por utilizar uma referência de método;

• linhas 45–51: certas exceções não passam pelas linhas 34–41. Isto acontece quando são feitas demasiadas solicitações. Assim que um determinado limite — que depende do ambiente de execução — é excedido, ocorre um [StackOverflowError], que é tratado pelas linhas 45–51;
• Linha 27: A subscrição produz um tipo [Subscription] que é adicionado a uma lista de subscrições. Esta lista terá apenas um elemento aqui;

Linha 32: Atualizamos a interface do utilizador utilizando o seguinte método [updateUi]:

    private void updateUi(UiResponse uiResponse) {
        // response time
        uiResponse.setResponseAt();
        // observation thread
        uiResponse.setObservedOn();
        // number of responses
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // running time
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // add the jSON response string to the JList response template
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

Aqui vemos que os componentes da interface gráfica do utilizador são atualizados (linhas 7, 9, 12). Para que isto seja possível, é necessário estar na thread da interface do utilizador (loop de eventos).

O método [endWaiting] é o seguinte:

    private void endWaiting() {
        // generate] button visible
        jButtonGenerate.setVisible(true);
        // hidden [Cancel] button
        jButtonCancel.setVisible(false);
        // hidden wait cursor
        jTabbedPane1.setCursor(Cursor.getDefaultCursor());
        // selected answers tab
        jTabbedPane1.setSelectedIndex(1);
        // duration updated one last time
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
}

O método [doCancel] é chamado quando ocorre um erro durante a execução de pedidos assíncronos ou quando o utilizador clica no botão [Cancelar]. O seu código é o seguinte:

// les souscriptions aux observables
    private List<Subscription> subscriptions = new ArrayList<Subscription>();
....
 
    @Override
    protected void doCancel() {
        // fin attente
        endWaiting();
        // dans le cas de souscriptions
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
            //subscriptions.forEach(s -> s.unsubscribe());
        }
    }
 

• linha 2: [subscriptions] é uma lista de subscrições;
• linha 11: todas as subscrições são canceladas;
• linha 12: outra forma de escrever a linha 11. O método [forEach] espera aqui uma instância do tipo Consumer<Subscription> (ver secção 4.4);

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

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

Se os observáveis utilizarem um dos agendadores [Schedulers.computation(), Scheduler.io(), Schedulers.newThread()], então serão executados fora da thread da interface do utilizador. Estas diferentes threads irão competir pelos núcleos da máquina. Uma vez que os pedidos são operações de longa duração (várias centenas de milissegundos), o método [doGenerateWithService] executado na thread da interface do utilizador terminará antes de os pedidos terem devolvido as suas respostas. No entanto, este método foi executado no evento de clique do botão [Generate]. Assim que este evento for processado, a thread da interface do utilizador poderá passar ao processamento dos eventos seguintes. Existem vários deles. Assim, o método [beginWaiting] tinha configurado vários:

    private void beginWaiting() {
        // buttons
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // waiting cursor
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // raz answers
        model.clear();
        // rx subscriptions
        subscriptions.clear();
        // the response view is displayed
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // start of execution
        debut = new Date().getTime();
}

Praticamente todas as linhas deste código afetam a interface gráfica do utilizador. Esta atualização não ocorre imediatamente: os eventos são colocados na fila do ciclo de eventos. Assim que o evento de clique no botão [Gerar] for processado, estes eventos são executados por ordem, e o utilizador pode ver a alteração na interface gráfica do utilizador:

• o separador [Resposta] é apresentado (linha 13) e um indicador de carregamento é associado a ele (linha 6)
• o seu botão [Cancel] é exibido (linha 4) e o utilizador pode clicar nele;
• a JList de respostas é esvaziada (linha 9);
• o JLabel para o número de respostas exibe 0;
• o JLabel para o tempo de execução exibe uma string vazia;

Ao longo da execução das consultas, a thread da interface do utilizador tem acesso regular ao processador. Pode então processar eventos pendentes. Entre estes estão os definidos pelo método [updateUi]:

    private void updateUi(UiResponse uiResponse) {
        // response time
        uiResponse.setResponseAt();
        // observation thread
        uiResponse.setObservedOn();
        // number of responses
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // running time
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // add the jSON response string to the JList response template
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

Quando a thread da interface do utilizador está ativa:

• o JLabel que exibe o número de respostas é atualizado (linha 7);
• o JLabel do tempo de execução é atualizado (linha 9);
• a JList de respostas é atualizada através do seu modelo (linha 12);

Isto permite ao utilizador ver o progresso da execução da consulta. Além disso, pode cancelá-la através do botão [Cancelar]. Este é o objetivo principal de ter serviços assíncronos na camada [Swing], e o RxJava é a tecnologia de eleição para a sua implementação.

Por fim, note que, se o utilizador escolher um dos agendadores [Schedulers.immediate(), Schedulers.trampoline()], os observáveis são então executados na mesma thread que o chamador, ou seja, a thread da interface do utilizador. Isto resulta num comportamento síncrono.

Os resultados obtidos com os diferentes agendadores foram apresentados nas secções 2.8.1, 2.8.2, 2.8.3 e 2.8.4.