RxJava, Programowanie Reaktywne - Wprowadzenie

Przyjrzymy się dzisiaj zyskującej coraz bardziej na popularności bibliotece JavaRX wprowadzającej programowanie reaktywne do Javy.

Programowanie reaktywne

Zacznijmy od krótkiego wprowadzenia. Programowanie reaktywne jest rozszerzeniem wzorca obserwatora. Mamy dwa główne typy:

• Observable- w dużym skrócie: dostarcza dane, zajmuje się również ich wcześniejszym przefiltrowaniem, mapowaniem etc.
• Observer - przetwarza dane, posiada 3 metody:
  • onNext() - normalne przetwarzanie elementu
  • onCompleted() - wywoływane po przetworzeniu wszystkich elementów
  • onError() - wywołane po wystąpieniu błędu

Działanie polega na przetwarzaniu danych w postaci strumieni. Observable dostarcza strumień danych, a Observer obsługuje jeden element po drugim. W przeciwieństwie do typowego wywoływania metod - nie wywołujemy metody i czekamy na odpowiedź, tylko czekamy i reagujemy (ang. react). Stąd właśnie nazwa tego podejścia.

Hello World

Zaczynamy od pobrania JARa i dodania go do projektu. Najnowszą wersję można znaleźć tutaj.

package pl.javastart.rxjava;

import rx.Observable;
import rx.functions.Action1;

public class App {

	public static void main(String[] names) {
		hello("World", "Kasia", "Tomek");
	}

	public static void hello(String... names) {
		Observable.from(names).subscribe(new Action1<String>() {

			@Override
			public void call(String s) {
				System.out.println("Hello " + s + "!");
			}
		});

	}
}

Wynik:

Hello World!
Hello Kasia!
Hello Tomek!

Kompatybilność z wyrażeniami lambda

RxJava jest kompatybilna z wyrażeniami lambda. Powyższy przykład możemy przekształcić na następujący

package pl.javastart.rxjava;

import rx.Observable;

public class App {

	public static void main(String[] names) {
		hello("Hello", "Kasia", "Tomek");
	}

	public static void hello(String... names) {
		Observable
		.from(names)
		.subscribe(s-> System.out.println("Hello " + s + "!"));
	}
}

Przykład

Poniżej przykład pełnej implementacji - wszystkich trzech metod obserwatora oraz wywołanie.

	void division() {
		Observer<Calc> obs =  new Observer<Calc>() {
            @Override
            public final void onNext(Calc calc) {
            	System.out.println(calc.a + "/" + calc.b + " = " + (calc.a / calc.b));
            }

            @Override
            public final void onCompleted() {
                System.out.println("Finished all");
            }

            @Override
            public final void onError(Throwable e) {
            	System.out.println("Error! " + e.getMessage());
            }
        };
		Calc[] calc = {new Calc(10,2), new Calc(10, 5), new Calc(5,2),  new Calc(100,1)}; 
		Observable.from(calc).subscribe(obs);
	}
	
	static class Calc {
		int a;
		int b;
		
		Calc(int a, int b) {
			this.a = a;
			this.b = b;
		}
	}

Wynik:

10/2 = 5
10/5 = 2
5/2 = 2
100/1 = 100
Finished all

Linki

• Repozytorium - https://github.com/ReactiveX/RxJava
• Prezentacja z wykorzystaniem w Androidzie - https://www.youtube.com/watch?v=JCLZ55M2gVo
• MVN Repository - http://mvnrepository.com/artifact/io.reactivex/rxjava