8. Swing ortamında RxJava

8.1. Introduction

Burada, 2. paragrafta tanıtılan Swing uygulamasına geri döneceğiz.

Swing ortamında RxJava ile çalışmak için, RxJava'e Swing ortamında kullanışlı sınıflar ve arayüzler ekleyen RxSwing kütüphanesini kullanacağız. Bunun için Swing örneğinin Gradle dosyası şu şekildedir:


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

15. satır: RxSwing'e bağımlılık;

RxSwing'e özgü tek bir nesne kullanacağız: Swing'in olay döngüsü iş parçacığı üzerinde gözlemlenebilirleri çalıştıran/gözlemleyen [SwingScheduler.getInstance()] zamanlayıcısı. Bunu, yalnızca olay döngüsünden farklı iş parçacıklarında çalışan gözlemlenebilirleri gözlemlemek için kullanacağız. Örnek uygulamanın mimarisini hatırlayalım:

Asenkron hizmet katmanı, gözlemlenebilirleri döndüren yöntemler içerir. Bu gözlemlenebilirleri, olay döngüsünden farklı iş parçacıklarında çalıştırıyoruz. Böylece grafik kullanıcı arayüzü donmaz ve kullanıcının taleplerine yanıt verebilir. En bariz örnek, kullanıcının çok uzun süren bir asenkron işlemi durdurmak için bir düğmeye ([Annuler]) tıklamasına izin vermektir. Bunun için tek şart, grafik arayüzün donmuş durumda olmamasıdır (frozen);
Swing katmanı, asenkron işlemler tarafından döndürülen sonuçları değerlendirmek ve bunlara dayanarak grafik arayüzü güncellemek ister. Ancak bu, yalnızca olay döngüsünün iş parçacığında yapılabilir. Bunun için bu sonuçlar [SwingScheduler.getInstance()] zamanlayıcısında izlenir;

Böylece, grafik arayüzün olay yönetimi kodunda, [rxService] asenkron katmanıyla etkileşim şu şekilde gerçekleşir:


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

burada [Schedulers.computation()] zamanlayıcısı, kullanım durumuna göre başka bir zamanlayıcıyla değiştirilebilir.

Okuyucunun 2. paragrafı tekrar okumasını öneririz. Artık bu paragrafı tam olarak anlayabilecek bilgiye sahiptir.

8.2. Kodun yapısı

Kod, aşağıdaki mimariyi uygular:

Bu mimariyi uygulayan IntelliJ IDEA projesi şöyledir:

[rxswing.service] paketi, senkron hizmet katmanlarını (IService, Service) ve asenkron hizmet katmanlarını (IRxService, RxService) uygular;
[rxswing.ui] paketi, Swing arayüzünü uygular;

8.3. Projenin çalıştırılması

Projeyi IntelliJ IDEA'da çalıştırmak için aşağıdaki adımları izleyin:

8.4. Senkron hizmet

Senkron hizmet katmanı, aşağıdaki [IService] arayüzünü sunar:


package dvp.rxswing.service;

public interface IService {
  // [a,b] aralığında rastgele sayılar
  // n sayıda sayı üretilir; burada n, [minCount, maxCount] aralığında rastgele bir sayıdır
  // sayılar, milisaniye gecikme süresinden sonra üretilir,
  // burada [delay], [minDelay, maxDelay] aralığında bir rastgele sayıdır
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

Hizmet yanıtının [ServiceResponse] türü şöyledir:


package dvp.rxswing.service;

import java.util.List;

public class ServiceResponse {

  // hizmetin bekleme süresi
  private int delay;
  // rastgele sayılar
  private List<Integer> aleas;
  // çalıştırma iş parçacığı
  private String executedOn;

  // yapıcılar

  public ServiceResponse() {
      // çalıştırma iş parçacığı
    executedOn = Thread.currentThread().getName();
  }

  public ServiceResponse(int delay, List<Integer> aleas) {
      // yerel oluşturucu
    this();
    // diğer başlatmalar
    this.delay = delay;
    this.aleas = aleas;
  }

  // alıcı ve ayarlayıcılar
...
}

[IService] arayüzü, aşağıdaki [Service] sınıfı tarafından uygulanır:


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) {
    // [a,b] aralığındaki rastgele sayılar
    // n sayıda sayı üretilir; burada n, [minCount, maxCount] aralığında rastgele bir sayıdır
    // sayılar, milisaniye gecikme süresinden sonra üretilir,
    // burada [delay], [minDelay, maxDelay] aralığında bir rastgele sayıdır

    // bazı kontroller
    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;
    }
    // hatalar mı?
    if (!messages.isEmpty()) {
      throw new AleasException(String.join(" [---] ", messages), erreur);
    }
    // rastgele sayı üreteci
    Random random = new Random();
    // bekleme?
    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);
      }
    }
    // sonuç üretme
    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));
    }
    // sonuç geri dönüşü
    return new ServiceResponse(delay,nombres);
  }

}

Hizmet tarafından kullanılan [AleasException] istisna sınıfı şöyledir:


package dvp.rxswing.service;

public class AleasException extends RuntimeException {

    private static final long serialVersionUID = 1L;
    // hata kodu
  private int code;

  // yapıcılar
  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;
  }

  // alıcı ve ayarlayıcılar
...
}

3. satır: [RuntimeException] sınıfını genişletir. Dolayısıyla bu, kontrol edilemeyen bir istisnadır;
7. satır: Üst sınıfına bir hata kodu ekler (0 = hata yok);

8.5. Asenkron hizmet

Asenkron hizmet katmanı, aşağıdaki [IRxService] arayüzünü sunar:


package dvp.rxswing.service;

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

public interface IRxService {
  // [a,b] aralığındaki rastgele sayılar
  // n sayıda sayı üretilir; burada n, [minCount, maxCount] aralığında rastgele bir sayıdır
  // sayılar, milisaniye gecikme süresinden sonra üretilir,
  // burada [delay], [minDelay, maxDelay] aralığında bir rastgele sayıdır
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

11. satır: Artık hizmetin [getAleas] yöntemi bir gözlemlenebilir değer döndürmektedir;

[getAleas] yöntemi, [Ui] katmanına yönelik [UiResponse] türünde bir yanıt döndürür. Bu tür şöyledir:


package dvp.rxswing.ui;

import dvp.rxswing.service.ServiceResponse;

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

public class UiResponse {

  // müşteri kimliği
  private int idClient;
  // hizmet yanıtı
  private ServiceResponse serviceResponse;
  // gözlem iş parçacığının adı
  private String observedOn;
  // istek saati
  private String requestAt;
  // yanıt saati
  private String responseAt;

  // yapıcılar

  public UiResponse() {
      // gözlem iş parçacığı
    observedOn = Thread.currentThread().getName();
    // istek saati
    requestAt = getTimeStamp();
  }

  // özel yöntemler

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

  // alıcı ve ayarlayıcılar
...
}

rastgele sayılar 13. satırdaki alanda yer alır;
diğer alanlar, asenkron hizmetin gözlemlenebilir öğesinin yürütme ve gözlem iş parçacıklarını ve hizmete yapılan isteğin ve alınan yanıtın saatlerini belirtmek için kullanılır;

Asenkron arayüz, aşağıdaki [RxService] sınıfı tarafından uygulanmaktadır:


package dvp.rxswing.service;

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

public class RxService implements IRxService {

  // senkron hizmet
  private IService service;

  // yapıcı
  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) {
      // senkron hizmet tarafından döndürülen değeri yayınlayan bir gözlemlenebilir oluşturulur
    return Observable.create(subscriber -> {
      try {
          // senkron çağrı
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        // sonuç gözlemciye aktarılır
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
          // hatayı gözlemciye aktarılır
        subscriber.onError(e);
      } finally {
          // gözlemciye yayınların bittiği bildirilir
        subscriber.onCompleted();
      }
    });
  }
}

12-14. satırlar: Asenkron hizmetin [RxService] sınıfı, senkron arayüz [IService]'in bir örneğinden oluşturulur;
20-33. satırlar: [getAleas] yönteminin sonucu olan gözlemlenebilirin oluşturulması;
satır 22: senkron [service.getAleas] yöntemi çağrılır. [ServiceResponse] türündeki sonucu, [swing] katmanına sağlanacak [UiResponse] türündeki nesneye dahil edilir. Bu nesne başlangıçta yöntemin çağrı parametreleri arasında aktarılmıştır (son parametre, satır 17);
24. satır: [UiResponse] yanıtı gözlemciye ([swing] katmanına) gönderilir. [UiResponse] nesnesi, yalnızca 22. satırdaki senkron hizmet tarafından oluşturulan bilgileri içermez. Aynı zamanda, 17. satırdaki [getAleas] yöntemini çağıran yöntem tarafından oluşturulan diğer bilgileri de içerir. Bu nedenle, bu çağıran yöntem [UiResponse] nesnesini [getAleas] yöntemine parametre olarak aktarmıştır (son parametre, satır 17);
satır 30: Yayınların bittiğini bildirmeyi unutmayalım. Burada, yalnızca tek bir değer yayan bir gözlemlenebilirimiz var: senkron hizmet tarafından döndürülen değer;
27. satır: Gözlemciye olası bir hata bildirilir;

8.6. Grafik arayüz

grafik arayüz, iyi bir grafik düzenleyiciye sahip olan IDE [Netbeans] ile oluşturulmuştur. Bu düzenleyici, yalnızca bu IDE tarafından kullanılabilen [AbstractJFrameAleas.form] dosyasını oluşturmuştur;
[AbstractJFrameAleas] sınıfı da NetBeans’in grafik düzenleyicisi tarafından oluşturulmuştur. Daha sonra şu şekilde yeniden yapılandırılmıştır: Yönetmek istediğimiz grafik arayüz olayları, [AbstractJFrameAleas] sınıfında, alt sınıf olan [JFrameAleasEvents] sınıfında uygulanan soyut yöntemler aracılığıyla işlenmektedir. Sonuç olarak,
- [AbstractJFrameAleas] soyut sınıfı, grafik kullanıcı arayüzünü oluşturmak ve görüntülemekle görevlidir;
- [JFrameAleasEvents] alt sınıfı, bu arayüzün olaylarını yönetmekle görevlidir;

[Request] sekmesinin grafik kullanıcı arayüzünün bileşenleri şunlardır:

no	tür	ad	işlev
1	JTabbedPane	jTabbedPane1	sekmeli bir konteyner. İki sekme içerir: (JPanel) [jPanelRequest] istek için, [jPanelresponse] yanıt için;
2	JTextField	jTextFieldNbValeurs	rastgele sayı hizmetine yapılacak istek sayısı. [Schedulers.io] zamanlayıcısında yürütülen asenkron hizmet durumunda, bu istekler bir işlemciyi paylaşacaktır;
3	JTextField	jTextFieldA	[a,b] aralığının a ucu
4	JTextField	jTextFieldB	aralığın b terminali [a,b]
5	JTextField	jTextFieldMinCount	[minCount, maxCount] aralığındaki minCount terminali
6	JTextField	jTextFieldMaxCount	[minCount, maxCount] aralığındaki maxCount terminali
7	JTextField	jTextFieldMinDelay	[minDelay, maxDelay] aralığındaki minDelay terminali
8	JTextField	jTextFieldMaxDelay	[minDelay, maxDelay] aralığındaki maxDelay terminali
9	JCheckBox	jCheckBoxRxSwing	Bu kutucuk işaretliyse, istekler asenkron arayüze gönderilir. Aksi takdirde, istekler senkron arayüze gönderilir
10	JComboBox	jComboBoxSchedulers	Asenkron istekler söz konusu olduğunda, bunlar burada seçilen zamanlayıcı ile yürütülür.
11	JButton	jButtonGenerate	İsteklerin yürütülmesini senkron veya asenkron hizmetine başlatır

[Response] sekmesinin grafik kullanıcı arayüzündeki bileşenler şunlardır:

no	tür	ad	işlev
1	JLabel	jLabelDuree	isteklerin toplam yürütme süresi (milisaniye cinsinden)
2	JLabel	jLabelNbReponses	gözlemlenen toplam yanıt sayısı (istek sayısından farklı olabilir; her istek birden fazla gözlem değeri sağlayabilir)
3	JList	jListNumbers	gözlemlenen (alınan) değerlerin gösterimi
4	JButton	jButtonAnnuler	çalışmakta olan istekleri iptal eder

8.7. Grafik kullanıcı arayüzünün başlatılması

[JFrameAleasEvents] sınıfı, grafik kullanıcı arayüzündeki olayları ve özellikle [Générer] düğmesine tıklamayı yönetir. Bu, aşağıdaki bağlamda çalışan bir yürütülebilir sınıftır:


public class JFrameAleasEvents extends AbstractJFrameAleas {

    private static final long serialVersionUID = 1L;
    // senkron üretim hizmeti
    private IService service;
    // asenkron üretim hizmeti
    private IRxService rxService;

    // girdiler
    private int nbRequests;
    private int a;
    private int b;
    private int minDelay;
    private int maxDelay;
    private int minCount;
    private int maxCount;

    // hata mesajları
    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 ";

    // gözlemlenebilir değişkenlere abonelikler
    protected List<Subscription> subscriptions = new ArrayList<Subscription>();
    // çalıştırma başlangıcı-bitişi
    private long debut;
    // jSON eşleştiricisi
    private ObjectMapper jsonMapper;
    // yanıt modeli
    private DefaultListModel<String> model;

    // oluşturucu
    public JFrameAleasEvents() {
        // üst öğe
        super();
        // yerel
        initJFrame();
        // hizmetler
        service = new Service();
        rxService = new RxService(service);
        // eşleştirici jSON
        jsonMapper = new ObjectMapper();
    }

    private void initJFrame() {
        // hata mesajlarını gizle
        jLabelCountError.setText("");
        jLabelDelayError.setText("");
        jLabelIntervalError.setText("");
        jLabelNbValuesError.setText("");
        // varsayılan metinler gizlenir
        jTextFieldA.setText("100");
        jTextFieldB.setText("200");
        jTextFieldMinCount.setText("5");
        jTextFieldMaxCount.setText("10");
        jTextFieldMinDelay.setText("100");
        jTextFieldMaxDelay.setText("500");
        jTextFieldNbValeurs.setText("10");
        jLabelDuree.setText("");
        // yanıt şablonu
        model = new DefaultListModel<>();
        jListNumbers.setModel(model);
        // çekirdek sayısı
        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);
        }

        /* Formu oluştur ve görüntüle */
        java.awt.EventQueue.invokeLater(() -> {
            new JFrameAleasEvents().setVisible(true);
        });
    }

1. satır: [JFrameAleasEvents] sınıfı, Swing sınıfı [JFrame]'i genişleten [AbstractJFrameAleas] sınıfını genişletir. Dolayısıyla [JFrameAleasEvents] sınıfı bir Swing penceresidir;
68-75. satırlar: Yürütülecek olan [main] yöntemi;
70. satır: Grafik kullanıcı arayüzünün görünüm ve hissiyatını belirler;
79. satır: [JFrameAleasEvents] sınıfının oluşturucusu çağrılır: grafik arayüz oluşturulacak ve başlatılacaktır. Bu işlem tamamlandıktan sonra arayüz görünür hale getirilir;
satır 34-44: oluşturucu;
satır 36: üst sınıfın oluşturucusunun çağrılması, grafik arayüzü başlatacaktır. Bu aşamada arayüz, geliştiricinin tasarladığı şekildedir. Henüz görünür değildir;
satır 38: grafik arayüzün bazı bileşenleri başlatılır;
satır 40: senkron hizmetin örneklenmesi;
satır 41: asenkron hizmetin örneklenmesi;

8.8. Senkron isteklerin yürütülmesi

[Générer] düğmesine tıklandığında, aşağıdaki [doGenerate] yöntemi yürütülür:


    @Override
    protected void doGenerate() {
        // girişler geçerli mi?
        if (!isPageValid()) {
            return;
        }
        // rx mi, değil mi?
        if (jCheckBoxRxSwing.isSelected()) {
            // asenkron istekler
            doGenerateWithRxService();
        } else {
            // senkron istekler
            doGenerateWithService();
        }
}

4-6. satırlar: Kullanıcının girdilerinin geçerli olup olmadığı kontrol edilir. [isPageValid] yöntemine değinmeyeceğiz. Bu yöntem basittir;
8. satır: RxSwing onay kutusunun durumu test edilir;
13. satır: istekler eşzamanlı olarak yürütülür;

[doGenerateWithService] yöntemi şu şekildedir:


    // senkron oluşturma
    private void doGenerateWithService() {
        // bekleme başlangıcı
        beginWaiting();
        try {
            for (int i = 0; i < nbRequests; i++) {
                // yanıt hazırlığı
                UiResponse uiResponse = new UiResponse();
                // müşteri numarası
                uiResponse.setIdClient(i);
                // senkron çağrı
                uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
                // yanıt saati
                uiResponse.setResponseAt();
                // JList şablonunun alınan yanıtlarla güncellenmesi
                model.add(0, jsonMapper.writeValueAsString(uiResponse));
                // yanıt sayısının güncellenmesi
                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);
        }
        // bekleme sonu
        endWaiting();
}

12. satır: rastgele sayı üretme hizmetine senkron çağrı yapılır;
[doGenerateWithService] yönteminin yürütülmesi tamamen Swing'in olay döngüsü iş parçacığında gerçekleşir. Yöntem tamamlanana kadar, grafik kullanıcı arayüzü hiçbir yeni olayı işlemez. Arayüz donmuş durumdadır (frozen). Örneğin, 16. ve 18. satırlardaki grafik arayüz güncellemeleri hiçbir zaman görülmeyecektir. Bunlar, ancak tüm isteklerin yürütülmesi tamamlandığında nihai değerleriyle birlikte görünür hale gelecektir;

[beginWaiting] yöntemi (4. satır) şu şekildedir:


    private void beginWaiting() {
        // düğmeler
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // bekleme imleci
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // 0 yanıt
        model.clear();
        // Rx abonelikleri
        subscriptions.clear();
        // yanıtlar ekrana yansıtılıyor
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // çalıştırma başlangıcı
        debut = new Date().getTime();
}

3. satır: [Générer] düğmesi gizlenmiştir. Bu, yine tüm isteklerin yürütülmesi tamamlandıktan sonra gerçekleştirilebilecek bir olay oluşturur. Ayrıca, [doGenerateWithService] yönteminin 25. satırındaki [endWaiting] yöntemi onu yeniden görüntülediği için, bu düğme hiçbir zaman gizli kalmaz;
13. satır: Gelen yanıtları görmek için [Response] sekmesini seçiyoruz. Yine, bu olay da ancak tüm isteklerin yürütülmesi tamamlandıktan sonra çalıştırılacak ve o zaman yanıtların tamamını göreceğiz, oysa biz yanıtların birbiri ardına gelmesini istiyorduk;

Senkron arayüzün açıkça yetersizlikleri vardır. Bunlar, asenkron arayüz sayesinde aşılmaktadır.

8.9. Asenkron isteklerin yürütülmesi

Asenkron isteklerin yürütülmesi için kod şu şekildedir:


private void doGenerateWithRxService() {
        // bekleme başlıyor
        beginWaiting();
        // rastgele sayıları bir gözlemlenebilir olarak alacağız
        Observable<UiResponse> observable = Observable.empty();
        // Farklı gözlemlenebilirlerin yürütme zamanlayıcısı
        Scheduler[] schedulers = { Schedulers.io(), Schedulers.computation(), Schedulers.newThread(),
                Schedulers.trampoline(), Schedulers.immediate() };
        Scheduler scheduler = schedulers[jComboBoxSchedulers.getSelectedIndex()];
        // gözlemlenebilirlerin yapılandırılması
        for (int i = 0; i < nbRequests; i++) {
            // yanıt hazırlığı
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // Gözlemlenebilir, kullanıcı tarafından seçilen zamanlayıcıda çalışacak şekilde yapılandırılır
            // ardından elde edilen gözlemlenebilirin genel gözlemlenebilirle birleştirilmesi
            observable = observable.mergeWith(
                    rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(scheduler));
        }
        // gözlemci
        observable = observable.observeOn(SwingScheduler.getInstance());
        // Şu an için sadece yapılandırma işlemi gerçekleştirildi
        // rastgele sayı üretme senkron hizmetine henüz hiçbir istek gönderilmedi
        // gözlemlenene abone olundu - bu, senkron rastgele sayı üretme hizmetinin çağrılmasına neden olacak
        try {
            // burada sadece bir abonelik var - sonuç bir abonelik oluşturulmasıdır
            subscriptions.add(observable.subscribe(
                    // yayın bildirimi
                    uiResponse -> {
                        // Yanıtla birlikte kullanıcı arayüzü güncellenir
                        // bu, gözlem işleminin kullanıcı arayüzü iş parçacığında gerçekleşmesi sayesinde mümkündür
                        updateUi(uiResponse);
                    } ,
                    // hata bildirimi
                    th -> {
                        // hata durumu - görüntüleniyor
                        String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
                        JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
                        // isteklerin iptal edilmesi
                        doCancel();
                    } ,
                    // bildirim [onCompleted]
                    // bekleme süresi sona erdi
                    this::endWaiting));
        } catch (Throwable th) {
            // istisna durumu + genel - görüntülenir
            String message = getInfoForThrowable("L'erreur suivante s'est produite", th);
            JOptionPane.showMessageDialog(this, message, "Informations", JOptionPane.PLAIN_MESSAGE);
            // istekler iptal edilir
            doCancel();
        }
    }

3. satır: Potansiyel olarak uzun sürecek bir işlemin devam ettiğini göstermek için grafik arayüz değiştirilir;
5. satır: Boş bir gözlemlenebilir nesne oluşturulur. [swing] katmanı tarafından gözlemlenecek olan bu gözlemlenebilir nesnedir;
7. satır: Olası zamanlayıcıların listesi;
9. satır: Kullanıcıya, isteklerin hangi zamanlayıcıda yürütüleceğini seçme olanağı verdik. Seçtiği zamanlayıcıyı alıyoruz;
satır 11-19: Her bir sorgu, elemanları 5. satırdaki gözlemlenebilir nesneye birleştirilen (mergeWith) (satır 17) bir gözlemlenebilir nesne döndürür;
satır 13-14: [UiResponse] nesnesi oluşturulur. Bu nesnenin hem [RxService.getAleas] yönteminin giriş parametresi hem de sonucu olduğunu hatırlatırız (satır 17-18);
14. satır: Her istek, burada [idClient] olarak adlandırılan numarasıyla tanımlanır. Bu, asenkron bir ortamda yanıtların alınma sırasının isteklerin gönderilme sırasından farklı olabileceği için gereklidir. [idClient], yanıtın hangi isteğe ait olduğunu belirlemeye olanak tanır;
satır 17-18: asenkron istek [rxService.getAleas] olarak oluşturulur. Bu istek, kullanıcı tarafından seçilen zamanlayıcı üzerinde yürütülür. Observable<UiResponse> türündeki sonucu, 5. satırdaki gözlemlenebilir ile birleştirilir. [rxService.getAleas] yönteminin burada yürütüldüğünü ve bir gözlemlenebilir döndürdüğünü iyi anlamak gerekir. Ancak bu, rastgele sayıların elde edildiği anlamına gelmez. Aslında, bir gözlemlenebilir ancak abone olunduğunda yürütülür. Henüz bu durum söz konusu değildir;
21. satır: Bu önemli bir komuttur: 5. satırdaki gözlemlenebilir nesne tarafından yayınlanan öğelerin, UI iş parçacığı üzerinde gözlemlenmesi talep edilmektedir. Burada, RxSwing kütüphanesine özgü bir zamanlayıcı kullanılmaktadır;
25-51. satırlar: 5. satırdaki gözlemlenebilir nesneye abone olunur. Ancak şimdi, rastgele sayılar bu sayıları üreten senkron hizmetten talep edilecektir. Önemli kısım 29-33. satırlardaki komutlardadır. Geri kalan kısım ise esas olarak hata durumlarını ve gözlemlenebilirin [onCompleted] bildirimini yönetir;
28-44. satırlar: 5. satırdaki işlemi UI iş parçacığında gözlemlemeyi talep ettiğimizi unutmamak gerekir. Dolayısıyla 28-44. satırlardaki kod, UI iş parçacığında yürütülür;
29-33. satırlar: Gözlemlenebilirin [onNext] bildirimini işliyoruz. Gözlemlenen işlem tarafından gönderilen bir [UiResponse] türü alıyoruz. Bu, asenkron isteklerden birinin sonucudur. Bu yanıtla grafik arayüzü güncelliyoruz;
satır 34-41: Gözlemlenebilir nesneden gelen [onError] bildirimini işliyoruz. Hatayı gösteren bir iletişim kutusu görüntülenir (satır 37-38), ardından istekler iptal edilir (satır 40);
satır 42-44: gözlemlenebilirin [onCompleted] bildirimini işliyoruz. İstenen hizmetin tamamlandığını göstermek için grafik arayüzü güncelliyoruz. Satır 44 şu şekilde de yazılabilirdi

 ()->{endWaiting();}

Burada bir yöntem referansı kullanmayı tercih ettik;

satır 45-51: Bazı istisnalar satır 34-41'den geçmez. Bu durum, çok fazla istek gönderildiğinde ortaya çıkar. Yürütme anındaki çalışma ortamına bağlı olarak belirli bir sınır aşıldığında, satır 45-51 tarafından yakalanan bir [StackOverflowError] oluşur;
27. satır: Abonelik, bir abonelik listesine eklenen [Subscription] türünde bir değer üretir. Bu liste burada tek bir öğe içerecektir;

32. satırda, aşağıdaki [updateUi] yöntemi ile grafik arayüz güncellenir:


    private void updateUi(UiResponse uiResponse) {
        // yanıt süresi
        uiResponse.setResponseAt();
        // gözlem iş parçacığı
        uiResponse.setObservedOn();
        // yanıt sayısı
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // çalışma süresi
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // yanıt dizesi jSON'in yanıt şablonu JList'e eklenmesi
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

Burada, grafik arayüzün bileşenlerinin güncellendiğini görüyoruz (7., 9. ve 12. satırlar). Bunun mümkün olabilmesi için, mutlaka Ui iş parçacığında (event loop) olunması gerekir.

[endWaiting] yöntemi şöyledir:


    private void endWaiting() {
        // [Générer] düğmesi görünür
        jButtonGenerate.setVisible(true);
        // [Annuler] düğmesi gizlendi
        jButtonCancel.setVisible(false);
        // bekleme imleci gizli
        jTabbedPane1.setCursor(Cursor.getDefaultCursor());
        // cevaplar sekmesi seçili
        jTabbedPane1.setSelectedIndex(1);
        // son güncelleme zamanı
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
}

[doCancel] yöntemi, asenkron isteklerin yürütülmesi sırasında bir hata oluştuğunda veya kullanıcı [Annuler] düğmesine tıkladığında çağrılır. Kod şöyledir:


// gözlemlenebilir öğelere abonelikler
    private List<Subscription> subscriptions = new ArrayList<Subscription>();
....

    @Override
    protected void doCancel() {
        // bekleme sonu
        endWaiting();
        // abonelikler söz konusu olduğunda
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
            //subscriptions.forEach(s -> s.unsubscribe());
        }
    }

2. satır: [subscriptions], bir abonelik listesidir;
11. satır: tüm abonelikler iptal edilir;
12. satır: 11. satırın başka bir yazımı. [forEach] yöntemi burada Consumer<Subscription> türünde bir örnek bekler (bkz. 4.4. paragraf);

[doGenerateWithService] yönteminin koduna geri dönelim: bu kod iki aşamaya ayrılabilir:

gözlemlenebilirlerin yapılandırılması aşaması. Bu, [doGenerateWithService] yöntemini çağıran iş parçacığında, yani kullanıcı arayüzünün iş parçacığında gerçekleştirilir;
gözlemlenebilirlerin yürütülmesini tetikleyecek abonelik;

Eğer gözlemlenebilirlerin zamanlayıcısı [Schedulers.computation(), Scheduler.io(), Schedulers.newThread()] zamanlayıcılarından biri ise, bunlar UI iş parçacığı dışında çalışacaktır. Bu farklı iş parçacıkları, makinenin bir veya birkaç çekirdeği için rekabet edecektir. İstekler uzun süren işlemler olduğundan (birkaç yüz milisaniye), UI iş parçacığında yürütülen [doGenerateWithService] yöntemi, istekler yanıtlarını döndürmeden önce tamamlanacaktır. Oysa bu yöntem, [Générer] düğmesine tıklama olayı üzerine çalıştırılmıştı. Bu olay işlendiğinden, UI iş parçacığı sonraki olayların işlenmesine geçebilecektir. Bunlardan birkaç tane vardır. Dolayısıyla, [beginWaiting] yöntemi birkaç tane ayarlamıştı:


    private void beginWaiting() {
        // düğmeler
        jButtonGenerate.setVisible(false);
        jButtonCancel.setVisible(true);
        // bekleme imleci
        jTabbedPane1.setCursor(Cursor.getPredefinedCursor(Cursor.WAIT_CURSOR));
        jButtonCancel.setCursor(Cursor.getDefaultCursor());
        // yanıtları sıfırla
        model.clear();
        // Rx abonelikleri
        subscriptions.clear();
        // yanıtlar görünümü görüntüleniyor
        jTabbedPane1.setSelectedIndex(1);
        jLabelNbReponses.setText("0");
        jLabelDuree.setText("");
        // çalıştırma başlangıcı
        debut = new Date().getTime();
}

Bu kodun hemen hemen her satırı, grafik arayüz üzerinde bir etkiye sahiptir. Bu güncelleme anında gerçekleşmez: olaylar, olay döngüsünün kuyruğuna yerleştirilir. [Générer] düğmesine tıklama olayı işlendiğinde, bu olaylar sırayla yürütülür ve kullanıcı grafik arayüzün değiştiğini görebilir:

[Response] sekmesi görüntülenir (13. satır) ve bu sekmeye bir bekleme imleci atanır (6. satır)
[Annuler] düğmesi görüntülenir (satır 4) ve kullanıcı bu düğmeye tıklayabilir;
yanıtlar için JList boşaltılır (satır 9);
yanıt sayısını gösteren JLabel, 0 değerini gösterir;
çalışma süresini gösteren JLabel, boş bir dize gösterir;

İsteklerin yürütülme süresi boyunca, UI iş parçacığı işlemciye düzenli olarak erişir. Böylece bekleyen olayları işleyebilir. Bunlar arasında, [updateUi] yöntemi tarafından ayarlananlar da vardır:


    private void updateUi(UiResponse uiResponse) {
        // yanıt saati
        uiResponse.setResponseAt();
        // gözlem dizisi
        uiResponse.setObservedOn();
        // yanıt sayısı
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
        // çalışma süresi
        jLabelDuree.setText(String.valueOf(new Date().getTime() - debut));
        // yanıt dizesi jSON'in yanıt şablonu JList'e eklenmesi
        try {
            model.add(0, jsonMapper.writeValueAsString(uiResponse));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
}

UI iş parçacığı kontrolü ele aldığında:

yanıt sayısını gösteren JLabel güncellenir (7. satır);
çalışma süresine ilişkin JLabel güncellenir (9. satır);
yanıtların JList değeri, şablonu aracılığıyla güncellenir (12. satır);

Böylece kullanıcı, isteklerin yürütülme sürecini takip edebilir. Ayrıca, [Annuler] düğmesi aracılığıyla bunları iptal edebilir. [swing] katmanının önünde asenkron hizmetlere sahip olmanın önemi işte budur ve RxJava, bunları uygulamak için tercih edilen bir teknolojidir.

Son olarak, kullanıcının [Schedulers.immediate(), Schedulers.trampoline()] zamanlayıcılarından birini seçmesi durumunda, gözlemlenebilirlerin çağrı yapanla aynı iş parçacığında, yani kullanıcı arayüzünün iş parçacığında yürütüldüğünü belirtelim. Bu durumda, sistem tekrar senkron çalışmaya döner.

Farklı zamanlayıcılarla elde edilen sonuçlar 2.8.1, 2.8.2, 2.8.3 ve 2.8.4 numaralı paragraflarda gösterilmiştir.