rx-java
Sujets

Syntaxe

• Sujet <T, R> subject = AsyncSubject.create (); // AsyncSubject par défaut
• Sujet <T, R> subject = BehaviorSubject.create (); // Comportement par défaut
• Sujet <T, R> subject = PublishSubject.create (); // PublishSubject par défaut
• Sujet <T, R> subject = ReplaySubject.create (); // ReplaySubject par défaut
• mySafeSubject = new SerializedSubject (unSafeSubject); // Convertit un unsafeSubject en un safeSubject - généralement pour les sujets multi-threadés

Paramètres

Remarques

Sujets de base

Un Subject dans RxJava est une classe à la fois Observable et Observer . Cela signifie essentiellement qu’il peut agir en tant Observable et transmettre des données aux abonnés et en tant Observer pour obtenir des données d’un autre observable.

Subject<String, String> subject = PublishSubject.create(); 
subject.subscribe(System.out::print);
subject.onNext("Hello, World!"); 

Les copies ci-dessus "Hello, World!" pour consoler en utilisant des Subjects .

Explication

1. La première ligne de code définit un nouveau Subject de type PublishSubject

   Subject<String, String> subject = PublishSubject.create();
       |     |       |       |                 |
    subject<input, output>  name   = default publish subject
   

2. La deuxième ligne s'abonne au sujet, montrant le comportement de l' Observer .

   subject.subscribe(System.out::print);
   

   Cela permet au Subject de prendre des entrées comme un abonné régulier

3. La troisième ligne appelle la méthode onNext du sujet, montrant le comportement Observable .

   subject.onNext("Hello, World!"); 
   

   Cela permet au Subject de donner des informations à tous les abonnés.

Les types

Un Subject (dans RxJava) peut être de l'un de ces quatre types:

• AsyncSubject
• Comportement
• PublishSubject
• ReplaySubject

De plus, un Subject peut être de type SerializedSubject . Ce type garantit que l' Subject ne viole pas le contrat observable (qui spécifie que tous les appels doivent être sérialisés)

Lectures complémentaires:

• Utiliser ou ne pas utiliser le sujet du blog de Dave Sexton

PublishSubject

PublishSubject n'émet sur un Observer que les éléments émis par l' Observable source après l'heure de l'abonnement.

Un exemple simple de PublishSubject :

Observable<Long> clock = Observable.interval(500, TimeUnit.MILLISECONDS);
Subject<Long, Long> subjectLong = PublishSubject.create();

clock.subscribe(subjectLong);

System.out.println("sub1 subscribing...");
subjectLong.subscribe(l -> System.out.println("sub1 -> " + l));
Thread.sleep(3000);
System.out.println("sub2 subscribing...");
subjectLong.subscribe(l -> System.out.println("sub2 -> " + l));
Thread.sleep(5000);

Sortie:

sub1 subscribing...
sub1 -> 0
sub1 -> 1
sub2 subscribing...
sub1 -> 2
sub2 -> 2
sub1 -> 3
sub2 -> 3

Dans l'exemple ci-dessus, un objet PublishSubject souscrit à une Observable qui agit comme une horloge et émet des éléments (Long) toutes les 500 millisecondes. Comme on le voit dans la sortie, le PublishSubject transmet les PublishSubject qu'il reçoit de la source ( clock ) à ses abonnés ( sub1 et sub2 ).

Un objet PublishSubject peut commencer à émettre des éléments dès qu'il est créé, sans aucun observateur, ce qui risque d'entraîner la perte d'un ou de plusieurs éléments jusqu'à ce qu'un observateur puisse s'inscrire.

createClock(); // 3 lines moved for brevity. same as above example

Thread.sleep(5000); // introduces a delay before first subscribe

sub1andsub2(); // 6 lines moved for brevity. same as above example

Sortie:

sub1 subscribing...
sub1 -> 10
sub1 -> 11
sub2 subscribing...
sub1 -> 12
sub2 -> 12
sub1 -> 13
sub2 -> 13

Notez que sub1 émet des valeurs à partir de 10 . Le délai de 5 secondes introduit a entraîné une perte d'articles. Ceux-ci ne peuvent pas être reproduits. Cela fait essentiellement de PublishSubject une Hot Observable .

Notez également que si un observateur s'abonne à PublishSubject après avoir émis n éléments, ces n éléments ne peuvent pas être reproduits pour cet observateur.

Ci-dessous le diagramme de marbre de PublishSubject

Le PublishSubject émet des éléments vers tous ceux qui ont souscrit, à tout moment avant l' onCompleted de onCompleted de la source Observable .

Si la source Observable termine par une erreur, le PublishSubject n'émettra aucun élément aux observateurs suivants, mais transmettra simplement la notification d'erreur de la source Observable.

Cas d'utilisation
Supposons que vous souhaitiez créer une application qui surveillera les cours des actions d’une certaine société et la transmettra à tous les clients qui en feront la demande.

/* Dummy stock prices */
Observable<Integer> prices = Observable.just(11, 12, 14, 11, 10, 12, 15, 11, 10);

/* Your server */
PublishSubject<Integer> watcher = PublishSubject.create();
/* subscribe to listen to stock price changes and push to observers/clients */
prices.subscribe(watcher);

/* Client application */
stockWatcher = getWatcherInstance(); // gets subject
Subscription steve = stockWatcher.subscribe(i -> System.out.println("steve watching " + i));
Thread.sleep(1000);
System.out.println("steve stops watching");
steve.unsubscribe();

Dans l'exemple ci-dessus, le PublishSubject fait office de passerelle pour transmettre les valeurs de votre serveur à tous les clients abonnés à votre watcher .

Lectures complémentaires:

• Publier des javadocs
• Blog de Thomas Nield (Lecture avancée)