rx-java
Soggetti

Sintassi

• Soggetto <T, R> subject = AsyncSubject.create (); // AsyncSubject predefinito
• Soggetto <T, R> subject = BehaviorSubject.create (); // Default BehaviorSubject
• Soggetto <T, R> subject = PublishSubject.create (); // Default PublishSubject
• Soggetto <T, R> subject = ReplaySubject.create (); // Default ReplaySubject
• mySafeSubject = new SerializedSubject (unSafeSubject); // Converti un unsafeSubject in un safeSubject - generalmente per soggetti multi-threaded

Parametri

Osservazioni

Soggetti di base

Un Subject in RxJava è una classe che è sia un Observable che un Observer . Ciò significa fondamentalmente che può agire come Observable e passare input agli abbonati e come Observer per ottenere input da un altro Observable.

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

Le stampe sopra "Hello, World!" consolare usando Subjects .

Spiegazione

1. La prima riga di codice definisce un nuovo Subject di tipo PublishSubject

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

2. La seconda riga si abbona al soggetto, mostrando il comportamento Observer .

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

   Ciò consente al Subject di prendere input come un normale abbonato

3. La terza riga chiama il metodo onNext dell'oggetto, mostrando il comportamento Observable .

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

   Ciò consente al Subject di fornire input a tutti coloro che lo sottoscrivono.

tipi

Un Subject (in RxJava) può essere di uno di questi quattro tipi:

• AsyncSubject
• BehaviorSubject
• PublishSubject
• ReplaySubject

Inoltre, un Subject può essere di tipo SerializedSubject . Questo tipo garantisce che il Subject non violi il Contratto Osservabile (che specifica che tutte le chiamate devono essere serializzate)

Ulteriori letture:

• Utilizzare o non usare Subject dal blog di Dave Sexton

PublishSubject

PublishSubject emette su un Observer solo quegli elementi che sono emessi dalla Observable origine successivamente al momento dell'abbonamento.

Un semplice esempio 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);

Produzione:

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

Nell'esempio precedente, PublishSubject iscrive a un Observable che agisce come un orologio ed emette elementi (Long) ogni 500 milli secondi. Come visto nell'output, PublishSubject passa i PublishSubject che ottiene dalla sorgente ( clock ) ai suoi abbonati ( sub1 e sub2 ).

Un PublishSubject può iniziare a emettere elementi non appena viene creato, senza alcun osservatore, che corre il rischio di perdere uno o più oggetti fino a quando un osservatore può oscurarsi.

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

Produzione:

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

Si noti che sub1 emette valori a partire da 10 . Il ritardo di 5 secondi introdotto ha causato una perdita di articoli. Questi non possono essere riprodotti. Questo rende essenzialmente PublishSubject un Hot Observable .

Si noti inoltre che se un osservatore si iscrive a PublishSubject dopo aver emesso n elementi, questi n elementi non possono essere riprodotti per questo osservatore.

Di seguito è riportato il diagramma di marmo di PublishSubject

PublishSubject emette elementi a tutti quelli che si sono iscritti, in qualsiasi momento prima che onCompleted l' onCompleted della Observable origine.

Se l' Observable origine termina con un errore, PublishSubject non emetterà alcun elemento agli osservatori successivi, ma passerà semplicemente la notifica di errore dalla Observable di origine.

Caso d'uso
Supponiamo che tu voglia creare un'applicazione che monitori i prezzi delle azioni di una certa azienda e inoltrala a tutti i clienti che ne fanno richiesta.

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

Nell'esempio di utilizzo sopra riportato, PublishSubject funge da bridge per trasferire i valori dal tuo server a tutti i client che si abbonano al tuo watcher .

Ulteriori letture:

• PublishSubject javadocs
• Blog di Thomas Nield (Lettura avanzata)