8. RxJava en el entorno Swing

8.1. Introducción

Aquí volveremos a examinar la aplicación Swing presentada en la sección 2.

Para trabajar con RxJava en un entorno Swing, utilizaremos la biblioteca RxSwing, que añade a RxJava clases e interfaces útiles en un entorno Swing. Para ello, el archivo Gradle del ejemplo de Swing es el siguiente:

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

• línea 15: la dependencia de RxSwing;

Solo utilizaremos un único objeto específico de RxSwing: el programador [SwingScheduler.getInstance()], que ejecuta u observa observables en el hilo del bucle de eventos de Swing. Lo utilizaremos exclusivamente para observar observables que se ejecutan en hilos distintos del bucle de eventos. Repasemos la arquitectura de la aplicación de ejemplo:

• la capa de servicio asíncrono proporciona métodos que devuelven observables. Ejecutamos estos observables en hilos distintos del hilo del bucle de eventos. De esta forma, la GUI sigue respondiendo. Puede reaccionar a las entradas del usuario. El ejemplo más obvio es permitir que el usuario haga clic en un botón [Cancelar] para interrumpir una operación asíncrona que está tardando demasiado. Para que esto funcione, la GUI debe estar congelada;
• la capa Swing necesita procesar los resultados devueltos por las operaciones asíncronas y utilizarlos para actualizar la interfaz gráfica de usuario. Sin embargo, esto solo se puede hacer en el hilo del bucle de eventos. Para lograrlo, estos resultados se observan en el programador [SwingScheduler.getInstance()];

Por lo tanto, en el código de gestión de eventos de la GUI, la interacción con la capa asíncrona [rxService] adopta la siguiente forma:

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

donde el programador [Schedulers.computation()] puede sustituirse por otro programador en función del caso de uso.

Se invita al lector a releer el párrafo 2. Ahora dispone de los conocimientos necesarios para comprenderlo plenamente.

8.2. La estructura del código

El código implementa la siguiente arquitectura:

El proyecto de IntelliJ IDEA que implementa esta arquitectura es el siguiente:

• el paquete [rxswing.service] implementa las capas de servicio síncronas (IService, Service) y asíncronas (IRxService, RxService);
• el paquete [rxswing.ui] implementa la interfaz Swing;

8.3. Ejecución del proyecto

Para ejecutar el proyecto en IntelliJ IDEA, sigue estos pasos:

8.4. El servicio síncrono

La capa de servicios síncronos proporciona la siguiente interfaz [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);
}

El tipo [ServiceResponse] de la respuesta del servicio es el siguiente:

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

La interfaz [IService] se implementa mediante la siguiente clase [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);
  }
 
}

La clase de excepción [AleasException] utilizada por el servicio es la siguiente:

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

• línea 3: extiende la clase [RuntimeException]. Por lo tanto, es una excepción no controlada;
• línea 7: añade un código de error a su clase padre (0 = sin error);

8.5. El servicio asíncrono

La capa de servicios asíncronos proporciona la siguiente interfaz [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);
}

• línea 11: el método [getAleas] del servicio ahora devuelve un observable;

El método [getAleas] devuelve una respuesta de tipo [UiResponse] destinada a la capa [Ui]. Este tipo es el siguiente:

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

• Los números aleatorios se encuentran en el campo de la línea 13;
• los demás campos sirven para especificar los subprocesos de ejecución y observación del observable del servicio asíncrono, así como las marcas de tiempo de la solicitud enviada al servicio y de la respuesta recibida;

La interfaz asíncrona se implementa mediante la siguiente clase [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();
      }
    });
  }
}

• líneas 12–14: la clase [RxService] del servicio asíncrono se construye a partir de una instancia de la interfaz síncrona [IService];
• líneas 20–33: construcción del observable, el resultado del método [getAleas];
• línea 22: se invoca el método síncrono [service.getAleas]. Su resultado, de tipo [ServiceResponse], se incluye en el objeto de tipo [UiResponse] que se va a proporcionar a la capa [swing]. Este objeto se pasó inicialmente en los parámetros de llamada del método (último parámetro, línea 17);
• línea 24: se envía el [UiResponse] al observador (la capa [swing]). El objeto [UiResponse] contiene no solo la información generada por el servicio síncrono en la línea 22, sino también otra información generada por el método que invoca al método [getAleas] en la línea 17. Por eso el método de llamada pasó el objeto [UiResponse] como parámetro al método [getAleas] (último parámetro, línea 17);
• línea 30: no nos olvidamos de señalar el final de las emisiones. Aquí tenemos un observable que emite un solo valor: el devuelto por el servicio síncrono;
• línea 27: notificamos al observador cualquier error;

8.6. La interfaz gráfica de usuario

• La interfaz gráfica de usuario se creó utilizando el IDE [NetBeans], que cuenta con un buen editor gráfico. Este editor generó el archivo [AbstractJFrameAleas.form], que solo puede utilizarse con este IDE;
• La clase [AbstractJFrameAleas] también fue generada por el editor gráfico de NetBeans. Posteriormente, se refactorizó de la siguiente manera: los eventos de la interfaz gráfica de usuario que queríamos gestionar se procesan en la clase [AbstractJFrameAleas] a través de métodos abstractos implementados en la clase hija [JFrameAleasEvents]. En definitiva,
  • la clase abstracta [AbstractJFrameAleas] se encarga de crear y mostrar la interfaz gráfica de usuario;
  • la clase hija [JFrameAleasEvents] gestiona sus eventos;

Los componentes de la interfaz gráfica de usuario de la pestaña [Request] son los siguientes:

N.º	tipo	nombre	función
1	JTabbedPane	jTabbedPane1	Un contenedor con pestañas. Contiene dos pestañas (JPanel): [jPanelRequest] para la solicitud y [jPanelResponse] para la respuesta;
2	JTextField	jTextFieldNbValues	el número de solicitudes que se realizarán al servicio de números aleatorios. En el caso del servicio asíncrono que se ejecuta en el programador [Schedulers.io], estas solicitudes compartirán un procesador;
3	JTextField	jTextFieldA	punto final a del intervalo [a,b]
4	JTextField	jTextFieldB	extremo b del intervalo [a,b]
5	JTextField	jTextFieldMinCount	minCount extremo del intervalo [minCount, maxCount]
6	JTextField	jTextFieldMaxCount	Límite superior del intervalo [minCount, maxCount]
7	JTextField	jTextFieldMinDelay	Límite minDelay del intervalo [minDelay, maxDelay]
8	JTextField	jTextFieldMaxDelay	Límite maxDelay del intervalo [minDelay, maxDelay]
9	JCheckBox	jCheckBoxRxSwing	Si la casilla está marcada, las solicitudes se envían a la interfaz asíncrona. De lo contrario, se envían a la interfaz síncrona
10	JComboBox	jComboBoxSchedulers	En el caso de las solicitudes asíncronas, se ejecutarán utilizando el programador seleccionado aquí
11	JButton	jButtonGenerate	inicia la ejecución de solicitudes al servicio síncrono o asíncrono

Los componentes de la interfaz gráfica de usuario de la pestaña [Respuesta] son los siguientes:

N.º	tipo	nombre	función
1	JLabel	jLabelDuration	el tiempo total de ejecución en milisegundos de las solicitudes
2	JLabel	jLabelNbResponses	el número total de respuestas observadas (puede diferir del número de solicitudes, ya que cada solicitud puede devolver varios valores que deben observarse)
3	JList	jListNumbers	visualización de los valores observados (recibidos)
4	JButton	jButtonCancel	cancela las solicitudes que se están ejecutando actualmente

8.7. Instanciación de la interfaz gráfica de usuario

La clase [JFrameAleasEvents] gestiona los eventos de la interfaz gráfica de usuario, incluidos los clics en el botón [Generar]. Se trata de una clase ejecutable que se ejecuta en el siguiente 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);
        });
    }

• línea 1: la clase [JFrameAleasEvents] hereda de la clase [AbstractJFrameAleas], que a su vez hereda de la clase [JFrame] de Swing. Por lo tanto, la clase [JFrameAleasEvents] es una ventana de Swing;
• líneas 68–75: el método [main] que se ejecutará;
• línea 70: establece el aspecto de la interfaz gráfica de usuario;
• línea 79: se invoca el constructor de la clase [JFrameAleasEvents]: se creará e inicializará la interfaz gráfica de usuario. Una vez hecho esto, se hace visible;
• líneas 34-44: el constructor;
• línea 36: la llamada al constructor padre inicializará la GUI. En este momento, tiene exactamente el aspecto que le dio el desarrollador. Todavía no es visible;
• línea 38: se inicializan ciertos componentes de la interfaz gráfica de usuario;
• línea 40: instanciación del servicio síncrono;
• línea 41: instanciación del servicio asíncrono;

8.8. Ejecución de solicitudes síncronas

Al hacer clic en el botón [Generar] se activa la ejecución del siguiente 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();
        }
}

• líneas 4–6: verificamos que la entrada del usuario sea válida. No comentaremos el método [isPageValid]. Es básico;
• línea 8: comprobamos el estado de la casilla de verificación RxSwing;
• línea 13: ejecutamos las solicitudes de forma sincrónica;

El método [doGenerateWithService] es el siguiente:

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

• línea 12: llamada sincrónica al servicio de generación de números aleatorios;
• el método [doGenerateWithService] se ejecuta íntegramente dentro del hilo del bucle de eventos de Swing. Hasta que el método no haya finalizado, la interfaz gráfica de usuario no procesa ningún evento nuevo. Queda congelada. Por lo tanto, por ejemplo, las actualizaciones de la interfaz gráfica de usuario de las líneas 16 y 18 nunca se verán. Solo serán visibles con sus valores finales, y esto ocurrirá al final de la ejecución de todas las solicitudes;

El método [beginWaiting] (línea 4) es el siguiente:

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

• línea 3: el botón [Generate] está oculto. Esto crea un evento que también solo se puede ejecutar una vez que todas las solicitudes hayan terminado de ejecutarse. Sin embargo, nunca lo vemos oculto, porque el método [endWaiting] de la línea 25 del método [doGenerateWithService] lo vuelve a mostrar;
• línea 13: seleccionamos la pestaña [Response] para ver llegar las respuestas. De nuevo, este evento solo se ejecutará una vez que todas las solicitudes hayan finalizado, momento en el que veremos todas las respuestas a la vez, mientras que queríamos verlas llegar una tras otra;

La interfaz síncrona tiene claras deficiencias. Estas se superan con la interfaz asíncrona.

8.9. Ejecución de solicitudes asíncronas

El código para ejecutar solicitudes asíncronas es el siguiente:

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

• línea 3: se actualiza la interfaz gráfica de usuario para indicar que se está llevando a cabo una operación que puede tardar bastante tiempo;
• línea 5: se crea un observable vacío. Este observable será observado por la capa [Swing];
• línea 7: la matriz de posibles programadores;
• línea 9: hemos dado al usuario la posibilidad de elegir el programador en el que ejecutar las consultas. Recuperamos el programador de su elección;
• líneas 11-19: cada solicitud devuelve un observable cuyos elementos se fusionan (mergeWith) (línea 17) con el observable de la línea 5;
• líneas 13-14: se construye el objeto [UiResponse]. Recordemos que este objeto es tanto el parámetro de entrada del método [RxService.getAleas] como su resultado (líneas 17-18);
• Línea 14: Cada solicitud se identifica mediante un número, denominado aquí [idClient]. Esto es necesario porque, en un entorno asíncrono, el orden en que se reciben las respuestas puede diferir del orden en que se enviaron las solicitudes. [idClient] nos permite determinar a qué solicitud pertenece la respuesta;
• Líneas 17-18: Se realiza la solicitud asíncrona [rxService.getRandom]. Se ejecuta en el programador elegido por el usuario. Su resultado, de tipo Observable<UiResponse>, se combina con el observable de la línea 5. Es importante señalar que el método [rxService.getAleas] se ejecuta aquí y devuelve un observable. Sin embargo, esto no significa que se hayan generado números aleatorios. De hecho, un observable solo se ejecuta cuando se suscribe a él. Este aún no es el caso;
• línea 21: esta es la instrucción clave: especificamos que los elementos emitidos por el observable de la línea 5 deben observarse en el hilo de la interfaz de usuario. Aquí utilizamos un programador específico de la biblioteca RxSwing;
• líneas 25–51: nos suscribimos al observable de la línea 5. Solo ahora se solicitarán los números aleatorios al servicio síncrono que los genera. La parte clave se encuentra en las instrucciones de las líneas 29–33. El resto se ocupa principalmente de los casos de error y de la notificación [onCompleted] del observable;
• líneas 28–44: Recuerde que solicitamos observar el proceso de la línea 5 en el hilo de la interfaz de usuario. Por lo tanto, el código de las líneas 28–44 se ejecuta en el hilo de la interfaz de usuario;
• líneas 29–33: Gestionamos la notificación [onNext] del observable. Recibimos un tipo [UiResponse] emitido por el proceso observado. Este es el resultado de una de las solicitudes asíncronas. Actualizamos la interfaz de usuario con esta respuesta;
• líneas 34–41: gestionamos la notificación [onError] del observable. Mostramos un cuadro de diálogo con el error (líneas 37–38) y, a continuación, cancelamos las solicitudes (línea 40);
• líneas 42–44: gestionamos la notificación [onCompleted] del observable. Actualizamos la interfaz gráfica de usuario para mostrar que el servicio solicitado se ha completado. La línea 44 también se podría haber escrito de la siguiente manera

 ()->{endWaiting();}

Aquí hemos optado por utilizar una referencia a un método;

• líneas 45–51: ciertas excepciones no pasan por las líneas 34–41. Esto ocurre cuando se realizan demasiadas solicitudes. Una vez que se supera un límite determinado —que depende del entorno de ejecución—, se produce un [StackOverflowError], que se gestiona en las líneas 45–51;
• Línea 27: La suscripción genera un tipo [Subscription] que se añade a una lista de suscripciones. Esta lista solo tendrá un elemento aquí;

Línea 32: Actualizamos la interfaz de usuario utilizando el siguiente 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();
        }
}

Aquí vemos que se actualizan los componentes de la interfaz gráfica de usuario (líneas 7, 9 y 12). Para que esto sea posible, debes estar en el hilo de la interfaz de usuario (bucle de eventos).

El método [endWaiting] es el siguiente:

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

El método [doCancel] se invoca cuando se produce un error durante la ejecución de solicitudes asíncronas o cuando el usuario hace clic en el botón [Cancelar]. Su código es el siguiente:

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

• línea 2: [suscripciones] es una lista de suscripciones;
• línea 11: se cancelan todas las suscripciones;
• línea 12: otra forma de escribir la línea 11. El método [forEach] espera aquí una instancia de tipo Consumer<Subscription> (véase la sección 4.4);

Volvamos al código del método [doGenerateWithService]: se puede dividir en dos pasos:

1. el paso de configuración de los observables. Esto se realiza en el hilo del llamante del método [doGenerateWithService], es decir, el hilo de la interfaz de usuario;
2. la suscripción que activará la ejecución de los observables;

Si los observables utilizan uno de los programadores [Schedulers.computation(), Scheduler.io(), Schedulers.newThread()], se ejecutarán fuera del hilo de la interfaz de usuario. Estos diferentes hilos competirán por los núcleos de la máquina. Dado que las solicitudes son operaciones de larga duración (varios cientos de milisegundos), el método [doGenerateWithService] ejecutado en el hilo de la interfaz de usuario finalizará antes de que las solicitudes hayan devuelto sus respuestas. Sin embargo, este método se había ejecutado al hacer clic en el botón [Generate]. Una vez procesado este evento, el hilo de la interfaz de usuario podrá pasar a procesar los siguientes eventos. Hay varios de ellos. Por lo tanto, el método [beginWaiting] había configurado varios:

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

Prácticamente cada línea de este código afecta a la interfaz gráfica de usuario. Esta actualización no se produce de forma inmediata: los eventos se colocan en la cola del bucle de eventos. Una vez procesado el evento de clic en el botón [Generar], estos eventos se ejecutan por turnos y el usuario puede ver cómo cambia la interfaz gráfica de usuario:

• se muestra la pestaña [Respuesta] (línea 13) y se le asocia un indicador de carga (línea 6)
• se muestra su botón [Cancel] (línea 4) y el usuario puede hacer clic en él;
• se borra la JList de respuestas (línea 9);
• el JLabel del número de respuestas muestra 0;
• el JLabel del tiempo de ejecución muestra una cadena vacía;

A lo largo de la ejecución de las consultas, el hilo de la interfaz de usuario tiene acceso regular al procesador. De este modo, puede procesar los eventos pendientes. Entre ellos se encuentran los establecidos por el 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();
        }
}

Cuando el hilo de la interfaz de usuario está activo:

• se actualiza el JLabel que muestra el número de respuestas (línea 7);
• se actualiza el JLabel del tiempo de ejecución (línea 9);
• se actualiza la JList de respuestas a través de su modelo (línea 12);

Esto permite al usuario ver el progreso de la ejecución de la consulta. Además, puede cancelarla mediante el botón [Cancelar]. Este es precisamente el objetivo de contar con servicios asíncronos delante de la capa [Swing], y RxJava es la tecnología elegida para implementarlos.

Por último, cabe señalar que si el usuario elige uno de los programadores [Schedulers.immediate(), Schedulers.trampoline()], los observables se ejecutan en el mismo hilo que el llamante, es decir, el hilo de la interfaz de usuario. Esto da lugar a un comportamiento síncrono.

Los resultados obtenidos con los diferentes programadores se mostraron en las secciones 2.8.1, 2.8.2, 2.8.3 y 2.8.4.