rx-java
Märkbar

Skapa en observerbar

Det finns flera sätt att skapa en observerbar i RxJava. Det mest kraftfulla sättet är att använda metoden Observable.create . Men det är också det mest komplicerade sättet . Så du måste undvika att använda det så mycket som möjligt.

Avger ett spännande värde

Om du redan har ett värde kan du använda Observable.just att avge ditt värde.

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

Avge ett värde som bör beräknas

Om du vill släppa ut ett värde som inte redan är beräknat, eller som kan ta lång tid att beräkna, kan du använda Observable.fromCallable att avge ditt nästa värde.

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

longComputation() kommer bara att longComputation() när du prenumererar på din Observable . På så sätt kommer beräkningen att vara lat .

Alternativt sätt att släppa ut ett värde som bör beräknas

Observable.defer bygger en Observable precis som Observable.fromCallable men den används när du behöver returnera en Observable istället för ett värde. Det är användbart när du vill hantera felen i ditt samtal.

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

Varma och kalla observabler

Observera är i stort sett kategoriserade som Hot eller Cold , beroende på deras utsläppsbeteende.
En Cold Observable är en som börjar sända på begäran (prenumeration), medan en Hot Observable är en som avger oberoende av prenumerationer.

Kallt observerbart

/* 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

Utgången från ovanstående kod ser ut (kan variera):

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

Observera att även om sub2 startar sent, får den värden från början. Avslutningsvis avger en Cold Observable endast artiklar när du begär det. Fler begäran startar flera pipelines.

Varmt observerbart

En Cold Observale kan konverteras till en Hot Observable med en enkel publish .

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

publish ger en ConnectableObservable som lägger till funktionaliteter för att ansluta och koppla bort från det observerbara.

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

Ovanstående ger:

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

Lägg märke till att även om sub2 börjar observera sent, är det synkroniserat med sub1 .
Koppla ur är lite mer komplicerat! Koppling sker på Subscription och inte på 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();

Ovanstående producerar:

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

Vid frånkoppling "avslutas" Observable huvudsak och startar om när ett nytt abonnemang läggs till.

Hot Observable kan användas för att skapa en EventBus . Sådana EventBussar är i allmänhet lätta och supersnabba. Den enda nackdelen med en RxBus är att alla händelser måste implementeras manuellt och skickas till bussen.