rx-java
Observable

Crear un observable

Hay varias formas de crear un Observable en RxJava. La forma más poderosa es usar el método Observable.create . Pero también es la forma más complicada . Así que debes evitar usarlo , tanto como sea posible.

Emitiendo un valor de salida.

Si ya tiene un valor, puede usar Observable.just para emitir su valor.

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

Emitiendo un valor que debe ser calculado.

Si desea emitir un valor que aún no está calculado, o que puede tardar mucho tiempo en calcularse, puede usar Observable.fromCallable para emitir su próximo valor.

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

longComputation() solo se llamará cuando te suscribas a tu Observable . De esta manera, el cálculo será perezoso .

Forma alternativa de emitir un valor que debe ser computado.

Observable.defer creado un Observable como Observable.fromCallable pero se usa cuando se necesita devolver un Observable lugar de un valor. Es útil cuando desea administrar los errores en su llamada.

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

Observables fríos y calientes

Los observables se clasifican ampliamente en Hot o Cold , dependiendo de su comportamiento de emisión.
Un Cold Observable es uno que comienza a emitir a pedido (suscripción), mientras que un Hot Observable es uno que emite independientemente de las suscripciones.

Frio observable

/* 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 salida del código anterior parece (puede variar):

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

Tenga en cuenta que aunque sub2 comienza tarde, recibe valores desde el principio. Para concluir, un Cold Observable solo emite elementos cuando se solicita. Múltiples solicitudes inician múltiples tuberías.

Caliente observable

Un Cold Observale se puede convertir en un Hot Observable con una publish simple.

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

publish devuelve un ConnectableObservable que agrega funcionalidades para conectarse y desconectarse de lo 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));

Los rendimientos anteriores:

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

Tenga en cuenta que aunque sub2 comienza a observar tarde, está sincronizado con sub1 .
¡Desconectar es un poco más complicado! La desconexión ocurre en la Subscription y no en el 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();

Lo anterior produce:

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

Al desconectarse, el Observable esencialmente "termina" y se reinicia cuando se agrega una nueva suscripción.

Hot Observable se puede utilizar para crear un EventBus . Tales EventBuses son generalmente ligeros y súper rápidos. El único inconveniente de un RxBus es que todos los eventos deben implementarse manualmente y pasarse al bus.