8. RxJava dans l'environnement Swing

8.1. Introduction

Nous allons ici revenir sur l'application Swing présentée au paragraphe 2.

Pour travailler avec RxJava dans un environnement Swing, nous utiliserons la bibliothèque RxSwing qui ajoute à RxJava des classes et interfaces utiles dans un environnement Swing. Pour cela, le fichier Gradle de l'exemple Swing est le suivant :

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'
}

• ligne 15 : la dépendance sur RxSwing ;

Nous n'utiliserons qu'un unique objet propre à RxSwing : le schéduler [SwingScheduler.getInstance()] qui fait exécuter / observer les observables sur le thread de l'event loop de Swing. Nous allons l'utiliser exclusivement pour observer des observables s'exécutant sur d'autres threads que celui l'event loop. Rappelons l'architecture de l'application exemple :

• la couche de service asynchrone présente des méthodes qui rendent des observables. Nous exécutons ces observables dans des threads différents de celui de l'event loop. Ainsi l'interface graphique n'est pas figée. Elle peut réagir à des sollicitations de l'utilisateur. La plus évidente est de permettre à celui-ci de cliquer sur un bouton [Annuler] pour interrompre une opération asynchrone trop longue. Pour qu'il puisse le faire, il ne faut que l'interface graphique soit figée (frozen) ;
• la couche Swing veut exploiter les résultats renvoyés par les opérations asynchrones et à partir d'eux mettre à jour l'interface graphique. Or ceci ne peut se faire que dans le thread de l'event loop. Pour ce faire, ces résultats sont observés dans le schéduler [SwingScheduler.getInstance()] ;

Ainsi dans le code de gestion des événements de l'interface graphique, l'interaction avec la couche asynchrone [rxService] se fait sous la forme suivante :

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

où le schéduler [Schedulers.computation()] pourra être remplacé par un autre schéduler selon les cas d'utilisation.

Le lecteur est invité à relire le paragraphe 2. Il a désormais les connaissances pour le comprendre complètement.

8.2. La structure du code

Le code implémente l'architecture suivante :

Le projet Intellij Idea implémentant cette architecture est le suivant :

• le package [rxswing.service] implémente les couches de service synchrones (IService, Service) et asynchrones (IRxService, RxService) ;
• le package [rxswing.ui] implémente l'interface Swing ;

8.3. Exécution du projet

Pour exécuter le projet sous Intellij Idea, procédez comme suit :

8.4. Le service synchrone

La couche de service synchrone présente l'interface [IService] suivante :

package dvp.rxswing.service;

public interface IService {
  // nombres aléatoires dans l'intervalle [a,b]
  // n nombres sont générés avec n lui-même nombre aléatoire dans l'intervalle [minCount, maxCount]
  // les nombres sont générés après une attente de delay millisecondes,
  // où [delay] est lui-même un nombre aléatoire dans l'intervalle [minDelay, maxDelay]
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

Le type [ServiceResponse] de la réponse du service est le suivant :

package dvp.rxswing.service;

import java.util.List;

public class ServiceResponse {

  // délai d'attente du service
  private int delay;
  // nombres aléatoires
  private List<Integer> aleas;
  // thread d'exécution
  private String executedOn;

  // constructeurs

  public ServiceResponse() {
      // thread d'exécution
    executedOn = Thread.currentThread().getName();
  }

  public ServiceResponse(int delay, List<Integer> aleas) {
      // constructeur local
    this();
    // autres initialisations
    this.delay = delay;
    this.aleas = aleas;
  }

  // getters et setters
...
}

L'interface [IService] est implémentée par la classe [Service] suivante :

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) {
    // nombres aléatoires dans l'intervalle [a,b]
    // n nombres sont générés avec n lui-même nombre aléatoire dans l'intervalle [minCount, maxCount]
    // les nombres sont générés après une attente de delay millisecondes,
    // où [delay] est lui-même un nombre aléatoire dans l'intervalle [minDelay, maxDelay]

    // qqs vérifications
    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;
    }
    // erreurs ?
    if (!messages.isEmpty()) {
      throw new AleasException(String.join(" [---] ", messages), erreur);
    }
    // générateur de nombres aléatoires
    Random random = new Random();
    // attente ?
    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);
      }
    }
    // génération résultat
    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));
    }
    // retour résultat
    return new ServiceResponse(delay,nombres);
  }

}

La classe d'exception [AleasException] utilisée par le service est la suivante :

package dvp.rxswing.service;

public class AleasException extends RuntimeException {

    private static final long serialVersionUID = 1L;
    // code d'erreur
  private int code;

  // constructeurs
  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 et setters
...
}

• ligne 3 : elle étend la classe [RuntimeException]. C'est donc une exception non contrôlée ;
• ligne 7 : elle enrichit sa classe parent d'un code d'erreur (0=pas d'erreur) ;

8.5. Le service asynchrone

La couche de service asynchrone présente l'interface [IRxService] suivante :

package dvp.rxswing.service;

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

public interface IRxService {
  // nombres aléatoires dans l'intervalle [a,b]
  // n nombres sont générés avec n lui-même nombre aléatoire dans l'intervalle [minCount, maxCount]
  // les nombres sont générés après une attente de delay millisecondes,
  // où [delay] est lui-même un nombre aléatoire dans l'intervalle [minDelay, maxDelay]
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

• ligne 11 : c'est désormais un observable que rend la méthode [getAleas] du service ;

La méthode [getAleas] rend une réponse de type [UiResponse] destinée à la couche [Ui]. Ce type est le suivant :

package dvp.rxswing.ui;

import dvp.rxswing.service.ServiceResponse;

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

public class UiResponse {

  // id du client
  private int idClient;
  // réponse du service
  private ServiceResponse serviceResponse;
  // nom du thread d'observation
  private String observedOn;
  // heure de la requête
  private String requestAt;
  // heure de la réponse
  private String responseAt;

  // constructeurs

  public UiResponse() {
      // thread d'observation
    observedOn = Thread.currentThread().getName();
    // heure de la requête
    requestAt = getTimeStamp();
  }

  // méthodes privées

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

  // getters et setters
...
}

• les nombres aléatoires sont dans le champ de la ligne 13 ;
• les autres champs sont là pour préciser les threads d'exécution et d'observation de l'observable du service asynchrone ainsi que les heures de la requête faite au service et de la réponse obtenue ;

L'interface asynchrone est implémentée par la classe [RxService] suivante :

package dvp.rxswing.service;

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

public class RxService implements IRxService {

  // service synchrone
  private IService service;

  // constructeur
  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) {
      // on crée un observable émettant la valeur rendue par le service synchrone
    return Observable.create(subscriber -> {
      try {
          // appel synchrone
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        // on passe le résultat à l'observateur
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
          // on passe l'erreur à l'observateur
        subscriber.onError(e);
      } finally {
          // on signale à l'observateur que les émissions sont finies
        subscriber.onCompleted();
      }
    });
  }
}

• lignes 12-14 : la classe [RxService] du service asynchrone est construite à partir d'une instance de l'interface synchrone [IService] ;
• lignes 20-33 : construction de l'observable, résultat de la méthode [getAleas] ;
• ligne 22 : la méthode synchrone [service.getAleas] est appelée. Son résultat de type [ServiceResponse] est inclus dans l'objet de type [UiResponse] à fournir à la couche [swing]. Cet objet a été initialement passé dans les paramètres d'appel de la méthode (dernier paramètre, ligne 17) ;
• ligne 24 : la réponse [UiResponse] est envoyée à l'observateur (la couche [swing]). L'objet [UiResponse] ne contient pas seulement les informations construites par le service synchrone ligne 22. Il contient également d'autres informations construites par la méthode appelante de la méthode [getAleas] de la ligne 17. C'est pour cette raison, que cette méthode appelante a passé l'objet [UiResponse] en paramètre à méthode [getAleas] (dernier paramètre, ligne 17) ;
• ligne 30 : on n'oublie pas de signaler la fin des émissions. On a là un observable n'émettant qu'une valeur : celle rendue par le service synchrone ;
• ligne 27 : on signale à l'observateur une éventuelle erreur ;

8.6. L'interface graphique

• l'interface graphique a été construite avec l'IDE [Netbeans] qui a un bon éditeur graphique. Cet éditeur a généré le fichier [AbstractJFrameAleas.form] exploitable seulement par cet IDE ;
• la classe [AbstractJFrameAleas] a été générée également par l'éditeur graphique de Netbeans. Elle a ensuite été refactorisée de la façon suivante : les événements de l'interface graphique qu'on voulait gérer sont traités dans la classe [AbstractJFrameAleas] par des méthodes abstraites implémentées dans la classe fille [JFrameAleasEvents]. Au final,
  • la classe abstraite [AbstractJFrameAleas] s'occupe de construire et d'afficher l'interface graphique ;
  • la classe fille [JFrameAleasEvents] s'occupe de gérer les événements de celle-ci ;

Les composants de l'interface graphique de l'onglet [Request] sont les suivants :

n°	type	nom	rôle
1	JTabbedPane	jTabbedPane1	un conteneur à onglets. Contient deux onglets (JPanel) [jPanelRequest] pour la requête, [jPanelresponse] pour la réponse ;
2	JTextField	jTextFieldNbValeurs	le nombre de requêtes à faire au service de nombres aléatoires. Dans le cas du service asynchrone exécuté sur le schéduler [Schedulers.io], ces requêtes vont se partager un processeur ;
3	JTextField	jTextFieldA	borne a de l'intervalle [a,b]
4	JTextField	jTextFieldB	borne b de l'intervalle [a,b]
5	JTextField	jTextFieldMinCount	borne minCount de l'intervalle [minCount, maxCount]
6	JTextField	jTextFieldMaxCount	borne maxCount de l'intervalle [minCount, maxCount]
7	JTextField	jTextFieldMinDelay	borne minDelay de l'intervalle [minDelay, maxDelay]
8	JTextField	jTextFieldMaxDelay	borne maxDelay de l'intervalle [minDelay, maxDelay]
9	JCheckBox	jCheckBoxRxSwing	si la case est cochée, les requêtes sont faites à l'interface asynchrone. Sinon elles sont faites à l'interface synchrone
10	JComboBox	jComboBoxSchedulers	dans le cas de requêtes asynchrones, celles-ci seront exécutées avec le schéduler choisi ici
11	JButton	jButtonGenerate	lance l'exécution des requêtes au service synchrone ou asynchrone

Les composants de l'interface graphique de l'onglet [Response] sont les suivants :

n°	type	nom	rôle
1	JLabel	jLabelDuree	la durée totale d'exécution en millisecondes des requêtes
2	JLabel	jLabelNbReponses	le nombre total de réponses observées (peut être différent du nombre de requêtes, chaque requête pouvant fournir plusieurs valeurs à observer)
3	JList	jListNumbers	affichage des valeurs observées (reçues)
4	JButton	jButtonAnnuler	annule les requêtes en cours d'exécution

8.7. Instanciation de l'interface graphique

La classe [JFrameAleasEvents] gère les événements de l'interface graphique et notamment le clic sur le bouton [Générer]. C'est une classe exécutable qui démarre dans le contexte suivant :

public class JFrameAleasEvents extends AbstractJFrameAleas {

    private static final long serialVersionUID = 1L;
    // service de génération synchrone
    private IService service;
    // service de génération asynchrone
    private IRxService rxService;

    // les saisies
    private int nbRequests;
    private int a;
    private int b;
    private int minDelay;
    private int maxDelay;
    private int minCount;
    private int maxCount;

    // messages d'erreur
    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 ";

    // les souscriptions aux observables
    protected List<Subscription> subscriptions = new ArrayList<Subscription>();
    // début-fin de l'exécution
    private long debut;
    // mappeur jSON
    private ObjectMapper jsonMapper;
    // model des réponses
    private DefaultListModel<String> model;

    // constructeur
    public JFrameAleasEvents() {
        // parent
        super();
        // local
        initJFrame();
        // services
        service = new Service();
        rxService = new RxService(service);
        // mappeur jSON
        jsonMapper = new ObjectMapper();
    }

    private void initJFrame() {
        // on cache les msg d'erreur
        jLabelCountError.setText("");
        jLabelDelayError.setText("");
        jLabelIntervalError.setText("");
        jLabelNbValuesError.setText("");
        // on cache les textes par défaut
        jTextFieldA.setText("100");
        jTextFieldB.setText("200");
        jTextFieldMinCount.setText("5");
        jTextFieldMaxCount.setText("10");
        jTextFieldMinDelay.setText("100");
        jTextFieldMaxDelay.setText("500");
        jTextFieldNbValeurs.setText("10");
        jLabelDuree.setText("");
        // model des réponses
        model = new DefaultListModel<>();
        jListNumbers.setModel(model);
        // nombre de coeurs
        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);
        });
    }

• ligne 1 : la classe [JFrameAleasEvents] étend la classe [AbstractJFrameAleas] qui étend elle-même la classe Swing [JFrame]. La classe [JFrameAleasEvents] est donc une fenêtre Swing ;
• lignes 68-75 : la méthode [main] qui va être exécutée ;
• ligne 70 : fixe le look and feel de l'interface graphique ;
• ligne 79 : le constructeur de la classe [JFrameAleasEvents] est appelé : l'interface graphique va être construite et initialisée. Ceci fait, elle est rendue visible ;
• lignes 34-44 : le constructeur ;
• ligne 36 : l'appel au constructeur parent va initialiser l'interface graphique. A ce moment là, elle est telle que le développeur l'a dessinée. Elle n'est pas encore visible ;
• ligne 38 : certains composants de l'interface graphique sont initialisés ;
• ligne 40 : instanciation du service synchrone ;
• ligne 41 : instanciation du service asynchrone ;

8.8. Exécution des requêtes synchrones

Le clic sur le bouton [Générer] provoque l'exécution de la méthode [doGenerate] suivante :

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

• lignes 4-6 : on vérifie que les saisies de l'utilisateur sont valides. Nous ne commenterons pas la méthode [isPageValid]. Elle est basique ;
• ligne 8 : on teste l'état de la case à cocher RxSwing ;
• ligne 13 : on exécute les requêtes de façon synchrone ;

La méthode [doGenerateWithService] est la suivante :

    // génération synchrone
    private void doGenerateWithService() {
        // début attente
        beginWaiting();
        try {
            for (int i = 0; i < nbRequests; i++) {
                // préparation réponse
                UiResponse uiResponse = new UiResponse();
                // n° client
                uiResponse.setIdClient(i);
                // appel synchrone
                uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
                // heure réponse
                uiResponse.setResponseAt();
                // mise à jour du modèle du JList avec les réponses reçues
                model.add(0, jsonMapper.writeValueAsString(uiResponse));
                // mise à jour du nombre de réponses
                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);
        }
        // fin attente
        endWaiting();
}

• ligne 12 : appel synchrone au service de génération des nombres aléatoires ;
• l'exécution de la méthode [doGenerateWithService] se fait entièrement dans le thread de l'event loop de Swing. Tant que la méthode n'est pas terminée, l'interface graphique ne traite aucun nouvel événement. Elle est figée (frozen). Ainsi par exemple, les mises à jour de l'interface graphique des lignes 16 et 18 ne seront jamais vues. Elles ne seront visibles qu'avec leurs valeurs finales et ceci à la fin de l'exécution de toutes les requêtes ;

La méthode [beginWaiting] (ligne 4) est la suivante :

    private void beginWaiting() {
        // boutons
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // curseur d'attente
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // raz réponses
        model.clear();
        // abonnements Rx
        subscriptions.clear();
        // on affiche la vue des réponses
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // début exécution
        debut = new Date().getTime();
}

• ligne 3 : le bouton [Générer] est caché. Cela crée un événement qui lui également ne pourra être exécuté qu'à la fin de l'exécution de toutes les requêtes. Aussi ne le voit-on jamais caché car la méthode [endWaiting] de la ligne 25 de la méthode [doGenerateWithService] l'affiche de nouveau ;
• ligne 13 : on sélectionne l'onglet [Response] pour voir les réponses arriver. De nouveau, cet événement ne sera exécuté qu'à la fin de l'exécution de toutes les requêtes où on verra alors la totalité des réponses alors qu'on voulait les voir arriver les unes après les autres ;

L'interface synchrone présente clairement des insuffisances. Celles-ci sont surmontées grâce à l'interface asynchrone.

8.9. Exécution des requêtes asynchrones

Le code d'exécution des requêtes asynchrones est le suivant :

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();
        }
    }

• ligne 3 : l'interface graphique est modifiée pour montrer qu'une opération potentiellement longue est en cours ;
• ligne 5 : on crée un observable vide. C'est cet observable qui sera observé par la couche [swing] ;
• ligne 7 : le tableau des schédulers possibles ;
• ligne 9 : nous avons donné la possibilité à l'utilisateur de choisir le schéduler sur lequel exécuter les requêtes. Nous récupérons le schéduler de son choix ;
• lignes 11-19 : chacune des requêtes rend un observable dont les éléments sont cumulés (mergeWith) (ligne 17) dans l'observable de la ligne 5 ;
• lignes 13-14 : l'objet [UiResponse] est construit. On rappelle que cet objet est à la fois paramètre d'entrée de la méthode [RxService.getAleas] et son résultat (lignes 17-18) ;
• ligne 14 : chaque requête est repérée par son numéro appelé ici [idClient]. Cela est nécessaire car dans un environnement asynchrone l'ordre de réception des réponses peut être différent de l'ordre d'émission des requêtes. [idClient] permet de savoir à quelle requête appartient la réponse ;
• lignes 17-18 : la requête asynchrone est faite [rxService.getAleas]. Elle est exécutée sur le schéduler choisi par l'utiliasteur. Son résultat de type Observable<UiResponse> est cumulé avec l'observable de la ligne 5. Il faut bien prendre conscience que la méthode [rxService.getAleas] est ici exécutée et rend un observable. Cela ne veut cependant pas dire que des nombres aléatoires ont été obtenus. En effet, un observable n'est exécuté que lorsqu'on s'y abonne. Ce n'est pas encore le cas ;
• ligne 21 : c'est l'instruction importante : on demande à ce que l'observation des éléments émis par l'observable de la ligne 5 soit faite sur le thread de l'Ui. On utilise ici un schéduler propre à la bibliothèque RxSwing ;
• lignes 25-51 : on s'abonne à l'observable de la ligne 5. Ce n'est que maintenant, que les nombres aléatoires vont être demandés au service synchrone de génération de ces nombres. L'essentiel est dans les instructions des lignes 29-33. Le reste gère essentiellement les cas d'erreurs et la notification [onCompleted] de l'observable ;
• lignes 28-44 : il faut se rappeler qu'on a demandé à observer le processus de la ligne 5 sur le thread de l'Ui. Donc le code des lignes 28-44 s'exécute dans le thread de l'Ui ;
• lignes 29-33 : on traite la notification [onNext] de l'observable. On reçoit un type [UiResponse] émis par le processus observé. C'est le résultat d'une des requêtes asynchrones. On met à jour l'interface graphique avec cette réponse ;
• lignes 34-41 : on traite la notification [onError] de l'observable. On affiche une boîte de dialogue affichant l'erreur (lignes 37-38) puis on annule les requêtes (ligne 40) ;
• lignes 42-44 : on traite la notification [onCompleted] de l'observable. On met à jour l'interface graphique pour montrer que le service demandé est terminé. La ligne 44 aurait également pu être écrite de la façon suivante

 ()->{endWaiting();}

On a préféré ici utiliser une référence de méthode ;

• lignes 45-51 : certaines exceptions ne passent pas par les lignes 34-41. C'est le cas, lorsqu'on demande trop de requêtes. Passée une certaine limite qui dépend de l'environnement de travail au moment de l'exécution, on a un [StackOverflowError] qui est intercepté par les lignes 45-51 ;
• ligne 27 : l'abonnement produit un type [Subscription] qui est ajouté à une liste de souscriptions. Celle-ci n'aura ici qu'un élément ;

Ligne 32, on met à jour l'interface graphique avec la méthode [updateUi] suivante :

    private void updateUi(UiResponse uiResponse) {
        // heure réponse
        uiResponse.setResponseAt();
        // thread d'observation
        uiResponse.setObservedOn();
        // nombre de réponses
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // durée d'exécution
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // ajout de la chaîne jSON de la réponse au modèle du JList des réponses
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

On voit ici que des composants de l'interface graphiques sont mis à jour (lignes 7, 9, 12). Pour que cela soit possible, il faut être obligatoirement dans le thread de l'Ui (event loop).

La méthode [endWaiting] est la suivante :

    private void endWaiting() {
        // bouton [Générer] visible
        jButtonGenerate.setVisible(true);
        // bouton [Annuler] caché
        jButtonCancel.setVisible(false);
        // curseur d'attente caché
        jTabbedPane1.setCursor(Cursor.getDefaultCursor());
        // onglet des réponses sélectionné
        jTabbedPane1.setSelectedIndex(1);
        // durée mise à jour uen ultime fois
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
}

La méthode [doCancel] est appelée lorsque survient une erreur dans l'exécution des requêtes asynchrones ou bien lorsque l'utilisateur clique sur le bouton [Annuler]. Son code est le suivant :

// 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());
        }
    }

• ligne 2 : [subscriptions] est une liste d'une souscription (abonnement) ;
• ligne 11 : toutes les souscriptions sont annulées ;
• ligne 12 : autre écriture de la ligne 11. La méthode [forEach] attend ici une instance de type Consumer<Subscription> (cf paragraphe 4.4) ;

Revenons au code de la méthode [doGenerateWithService] : il peut être décomposé en deux étapes :

1. étape de configuration des observables. Cela se fait dans le thread de l'appelant de la méthode [doGenerateWithService], ç-à-d le thread de l'Ui ;
2. l'abonnement qui va provoquer l'exécution des observables ;

Si les observables ont pour schéduler, l'un des schédulers [Schedulers.computation(), Scheduler.io(), Schedulers.newThread()], alors ils vont s'exécuter en-dehors du thread de l'Ui. Ces différents threads vont se disputer le ou les coeurs de la machine. Comme les requêtes sont des opérations longues (plusieurs centaines de millisecondes), la méthode [doGenerateWithService] exécutée dans le thread de l'Ui va se terminer avant que les requêtes n'aient renvoyé leurs réponses. Or cette méthode avait été exécutée sur l'événement clic sur le bouton [Générer]. Cet événement étant traité, le thread de l'Ui va pouvoir passer au traitement des événements suivants. Il y en a plusieurs. Ainsi, la méthode [beginWaiting] en avait positionné plusieurs :

    private void beginWaiting() {
        // boutons
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // curseur d'attente
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // raz réponses
        model.clear();
        // abonnements Rx
        subscriptions.clear();
        // on affiche la vue des réponses
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // début exécution
        debut = new Date().getTime();
}

Pratiquement toutes les lignes de ce code ont une action sur l'interface graphique. Cette mise à jour n'a pas lieu immédiatement : des événements sont placés dans la file d'attente de l'event loop. Une fois l'événement clic sur le bouton [Générer] traité, ces événements sont exécutés à leur tour et l'utilisateur peut voir l'interface graphique changer :

• l'onglet [Response] est affiché (ligne 13) et on lui associe un curseur d'attente (ligne 6)
• son bouton [Annuler] est affiché (ligne 4) et l'utilisateur pourra cliquer dessus ;
• le JList des réponses est vidé (ligne 9) ;
• le JLabel du nombre de réponses affiche 0 ;
• le JLabel de la durée d'exécution affiche une chaîne vide ;

Pendant toute la durée d'exécution des requêtes, le thread de l'UI a régulièrement accès au processeur. Il peut alors traiter les événements en attente. Parmi ceux-ci, il y a ceux positionnés par la méthode [updateUi] :

    private void updateUi(UiResponse uiResponse) {
        // heure réponse
        uiResponse.setResponseAt();
        // thread d'observation
        uiResponse.setObservedOn();
        // nombre de réponses
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // durée d'exécution
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // ajout de la chaîne jSON de la réponse au modèle du JList des réponses
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

Lorsque le thread de l'Ui a la main :

• le JLabel du nombre de réponses est mis à jour (ligne 7) ;
• le JLabel de la durée d'exécution est mis à jour (ligne 9) ;
• le JList des réponses est mis à jour au travers de son modèle (ligne 12) ;

Ainsi l'utilisateur voit la progression de l'exécution des requêtes. Par ailleurs, il peut les annuler via le bouton [Annuler]. C'est là tout l'intérêt d'avoir des services asynchrones en face de la couche [swing] et RxJava est une technologie de choix pour implémenter ceux-ci.

Pour terminer, notons que si l'utilisateur choisit l'un des schédulers [Schedulers.immediate(), Schedulers.trampoline()], les observables sont alors exécutés sur le même thread que l'appelant, ç-à-d le thread de l'Ui. On est alors ramené à un fonctionnement synchrone.

Les résultats obtenus avec les différents schédulers ont été montrés aux paragraphes 2.8.1, 2.8.2, 2.8.3, 2.8.4.