rx-java
Observable

Créer une observable

Il existe plusieurs façons de créer une observable dans RxJava. Le moyen le plus efficace consiste à utiliser la méthode Observable.create . Mais c'est aussi la manière la plus compliquée . Vous devez donc éviter de l'utiliser autant que possible.

Émettre une valeur existante

Si vous avez déjà une valeur, vous pouvez utiliser Observable.just pour émettre votre valeur.

 Observable.just("Hello World").subscribe(System.out::println);

Emettre une valeur qui devrait être calculée

Si vous voulez émettre une valeur qui n'est pas déjà calculée ou qui peut prendre du temps à être calculée, vous pouvez utiliser Observable.fromCallable pour émettre votre prochaine valeur.

Observable.fromCallable(() -> longComputation()).subscribe(System.out::println);

longComputation() ne sera appelé que lorsque vous vous abonnez à votre Observable . De cette façon, le calcul sera paresseux .

Autre manière d'émettre une valeur à calculer

Observable.defer construit un Observable tout comme Observable.fromCallable mais il est utilisé lorsque vous devez retourner un Observable au lieu d'une valeur. C'est utile quand vous voulez gérer les erreurs dans votre appel.

Observable.defer(() -> {
          try {
                return Observable.just(longComputation());
          } catch(SpecificException e) {
                return Observable.error(e);
          }).subscribe(System.out::println);

Observables chauds et froids

Les observables sont généralement classés comme Hot ou Cold , selon leur comportement en matière d’émission.
Un objet Cold Observable est celui qui commence à émettre sur demande (abonnement), tandis qu'un objet Hot Observable est celui qui émet indépendamment des abonnements.

Observable à froid

/* Demonstration of a Cold Observable */
Observable<Long> cold = Observable.interval(500, TimeUnit.MILLISECONDS); // emits a long every 500 milli seconds
cold.subscribe(l -> System.out.println("sub1, " + l)); // subscriber1
Thread.sleep(1000); // interval between the two subscribes
cold.subscribe(l -> System.out.println("sub2, " + l)); // subscriber2

La sortie du code ci-dessus ressemble (peut varier):

sub1, 0    -> subscriber1 starts
sub1, 1
sub1, 2
sub2, 0    -> subscriber2 starts
sub1, 3
sub2, 1
sub1, 4
sub2, 2

Notez que même si sub2 commence en retard, il reçoit des valeurs dès le début. Pour conclure, un Cold Observable n'émet que des éléments demandés. Une demande multiple démarre plusieurs pipelines.

Observable à chaud

Une Cold Observale peut être convertie en une Hot Observable avec une simple publish .

Observable.interval(500, TimeUnit.MILLISECONDS)
    .publish(); // publish converts cold to hot

publish renvoie un objet ConnectableObservable qui ajoute des fonctionnalités pour se connecter et se déconnecter de l'observable.

ConnectableObservable<Long> hot = Observable
                                    .interval(500, TimeUnit.MILLISECONDS)
                                    .publish(); // returns ConnectableObservable
hot.connect(); // connect to subscribe

hot.subscribe(l -> System.out.println("sub1, " + l));
Thread.sleep(1000);
hot.subscribe(l -> System.out.println("sub2, " + l));

Les rendements ci-dessus:

sub1, 0  -> subscriber1 starts
sub1, 1
sub1, 2
sub2, 2  -> subscriber2 starts
sub1, 3
sub2, 3

Notez que même si sub2 commence à observer tard, il est synchronisé avec sub1 .
Déconnecter est un peu plus compliqué! La déconnexion se produit sur l' Subscription et non sur l' Observable .

ConnectableObservable<Long> hot = Observable
                                    .interval(500, TimeUnit.MILLISECONDS)
                                    .publish(); // same as above
Subscription subscription = hot.connect(); // connect returns a subscription object, which we store for further use

hot.subscribe(l -> System.out.println("sub1, " + l));
Thread.sleep(1000);
hot.subscribe(l -> System.out.println("sub2, " + l));
Thread.sleep(1000);
subscription.unsubscribe(); // disconnect, or unsubscribe from subscription

System.out.println("reconnecting");
/* reconnect and redo */
subscription = hot.connect();
hot.subscribe(l -> System.out.println("sub1, " + l));
Thread.sleep(1000);
hot.subscribe(l -> System.out.println("sub2, " + l));
Thread.sleep(1000);
subscription.unsubscribe();

Le ci-dessus produit:

sub1, 0   -> subscriber1 starts
sub1, 1
sub1, 2
sub2, 2   -> subscriber2 starts
sub1, 3
sub2, 3
reconnecting  -> reconnect after unsubscribe
sub1, 0
...

Lors de la déconnexion, l' Observable "termine" essentiellement et redémarre lorsqu'un nouvel abonnement est ajouté.

Hot Observable peut être utilisé pour créer un EventBus . Ces EventBuses sont généralement légers et super rapides. Le seul inconvénient d'un RxBus est que tous les événements doivent être implémentés manuellement et transmis au bus.