rx-java
Asignaturas

Sintaxis

• Asunto <T, R> subject = AsyncSubject.create (); // AsyncSubject predeterminado
• Asunto <T, R> subject = BehaviorSubject.create (); // Default BehaviorSubject
• Asunto <T, R> subject = PublishSubject.create (); // Predeterminado PublishSubject
• Asunto <T, R> subject = ReplaySubject.create (); // ReplaySubject predeterminado
• mySafeSubject = new SerializedSubject (unSafeSubject); // Convertir un unsafeSubject en un safeSubject - generalmente para temas de múltiples hilos

Parámetros

Observaciones

Materias basicas

Un Subject en RxJava es una clase que es tanto un Observable como un Observer . Básicamente, esto significa que puede actuar como un Observable y pasar entradas a los suscriptores y como un Observer para obtener entradas de otro Observable.

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

Las impresiones de arriba "¡Hola mundo!" consolar utilizando Subjects .

Explicación

1. La primera línea de código define un nuevo Subject de tipo PublishSubject

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

2. La segunda línea se suscribe al sujeto, mostrando el comportamiento del Observer .

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

   Esto permite al Subject tomar entradas como un suscriptor regular

3. La tercera línea llama al método onNext del sujeto, mostrando el comportamiento Observable .

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

   Esto permite al Subject dar entradas a todos los que se suscriban.

Los tipos

Un Subject (en RxJava) puede ser de cualquiera de estos cuatro tipos:

• AsyncSubject
• Comportamiento
• PublishSubject
• ReplaySubject

Además, un Subject puede ser de tipo SerializedSubject . Este tipo garantiza que el Subject no infringe el contrato observable (que especifica que todas las llamadas deben ser serializadas)

Otras lecturas:

• Para usar o no usar el tema del blog de Dave Sexton

PublishSubject

PublishSubject emite a un Observer solo aquellos elementos que emite la fuente Observable después de la suscripción.

Un ejemplo 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);

Salida:

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

En el ejemplo anterior, un PublishSubject suscribe a un Observable que actúa como un reloj y emite elementos (Largo) cada 500 mili segundos. Como se ve en la salida, PublishSubject pasa los PublishSubject que recibe de la fuente ( clock ) a sus suscriptores ( sub1 y sub2 ).

Un PublishSubject puede comenzar a emitir elementos tan pronto como se crea, sin ningún observador, lo que corre el riesgo de perder uno o más elementos hasta que un observador pueda suscribirse.

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

Salida:

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

Observe que sub1 emite valores a partir de 10 . El retraso de 5 segundos introducido causó una pérdida de artículos. Estos no pueden ser reproducidos. Esto esencialmente hace de PublishSubject un Hot Observable .

Además, tenga en cuenta que si un observador se suscribe a PublishSubject después de que haya emitido n elementos, estos n elementos no se pueden reproducir para este observador.

A continuación se muestra el diagrama de mármol de PublishSubject

PublishSubject emite elementos a todos los que se han suscrito, en cualquier momento antes de que se onCompleted al onCompleted de la fuente Observable .

Si la fuente Observable termina con un error, PublishSubject no emitirá ningún elemento a los observadores posteriores, sino que simplemente transmitirá la notificación de error desde la fuente observable.

Caso de uso
Supongamos que desea crear una aplicación que supervise los precios de las acciones de una determinada empresa y la envíe a todos los clientes que la soliciten.

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

En el caso de uso de ejemplo anterior, PublishSubject actúa como un puente para pasar los valores de su servidor a todos los clientes que se suscriben a su watcher .

Otras lecturas:

• PublishSubject javadocs
• Blog de Thomas Nield (Lectura avanzada)