@FMuntenescu

Florina Muntenescu@FMuntenescu

Building a Reactive Mindset

@FMuntenescu

Florina Muntenescu@FMuntenescu

Building a Reactive Mindset

@FMuntenescu

Donald

Duck’s Burger

Joint

@FMuntenescu

Huey Dewey Louie

@FMuntenescu

Huey

@FMuntenescu

Dewey

@FMuntenescu

Louie

@FMuntenescu

Huey

Dewey

Louie

@FMuntenescu

filter(meat -> isMeatFresh(meat))

map(meat -> cook(meat))

Dewey

@FMuntenescu

zip

@FMuntenescu

zip

@FMuntenescu

zip

@FMuntenescu

zip

@FMuntenescuSTREAMS OF DATA

@FMuntenescuSTREAMS OF DATA

@FMuntenescuSTREAMS OF DATA

@FMuntenescu

zip

STREAMS OF DATA

@FMuntenescu

Consumer

Observable

Subscriber

@FMuntenescu

filter(meat -> isMeatFresh(meat))

map(meat -> cook(meat))

operators

@FMuntenescu

Huey.getBun()

@FMuntenescu

filter(meat -> isMeatFresh(meat))

for(Meat meat: meatList){

if(!isMeatFresh(meat)){

meatList.remove(meat);

}

}

@FMuntenescu

Observable.from(meatSource)

.filter(meat ->

isMeatFresh(meat))

filter(meat -> isMeatFresh(meat))

@FMuntenescu

filter(meat -> isMeatFresh(meat))

map(meat -> cook(meat))

for(Meat meat: meatList){

meat = cook(meat);

}

@FMuntenescu

Observable.from(meatSource)

.filter(meat ->

isMeatGoneBad(meat))

.map(meat -> cook(meat));

filter(meat -> isMeatFresh(meat))

map(meat -> cook(meat))

@FMuntenescu

Observable<Burger> burgerObservable =

Observable.zip(bunObservable,

meatObservable,

tomatoObservable,

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

bunObservable

meatObservable

tomatoObservable

burgerObservable

@FMuntenescu

Observable<Burger> burgerObservable =

Observable.zip(bunObservable,

meatObservable,

tomatoObservable,

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

bunObservable

meatObservable

tomatoObservable

burgerObservable

@FMuntenescuConsumer

Observable

Subscriber

.subscribe

@FMuntenescu

Subscriber

burgerObservable

.subscribe(mickeySubscriber);

@FMuntenescu

Subscriber

onNext(Burger burger)

onCompleted()

burgerObservable

.subscribe(

burger -> eatBurger(burger),

error -> complain(error),

() -> leaveBurgerJoint());

// onNext

//onError

//onCompleted

onError(Throwable error)

@FMuntenescu

Observable.zip(bunObservable,

meatObservable,

tomatoObservable,

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

.subscribe(burger -> eatBurger(burger),

error -> complain(error));

// dining area

// dining area

@FMuntenescu

// kitchen

Observable.zip(bunObservable,

meatObservable,

tomatoObservable,

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

.subscribe(burger -> eatBurger(burger),

error -> complain(error));

// dining area

@FMuntenescu

// kitchen

// dining area

Observable.zip(bunObservable,

meatObservable,

tomatoObservable,

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

.subscribeOn(kitchen)

.observeOn(dining area)

.subscribe(burger -> eatBurger(burger),

error -> complain(error));

@FMuntenescu

Observable.zip(bunObservable,

meatObservable,

tomatoObservable,

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

.subscribeOn(Schedulers.computation())

.observeOn(AndroidSchedulers.mainThread())

.subscribe(burger -> eatBurger(burger),

error -> complain(error));

// kitchen

// dining area

@FMuntenescu

Stream of data

ObservableSubscriber

OperatorsThreads

@FMuntenescu

Creating

Observables

@FMuntenescu

Creating

Observables

From a

List

@FMuntenescu

List<Burger> burgers = new ArrayList<>();

burgers.add(new Burger());

burgers.add(new Burger());

Observable<Burger> burgerObservable =

Observable.from(burgers);

@FMuntenescu

List<Burger> burgers = new ArrayList<>();

burgers.add(new Burger());

burgers.add(new Burger());

Observable<Burger> burgerObservable =

Observable.from(burgers);

1 2

@FMuntenescu

Observable<Burger> burgerObservable =

Observable.from(burgers);

burgerObservable.subscribe(new Subscriber<Burger>() {

@Override

public void onNext(final Burger burger) {

}

@Override

public void onCompleted() {

}

@Override

public void onError(final Throwable e) {

}

});

@FMuntenescu

Observable<Burger> burgerObservable = Observable.from(burgers);

burgerObservable.subscribe(new Subscriber<Burger>() {

@Override

public void onNext(final Burger burger) {

// handle burger 1

}@Override

public void onCompleted() {

}

@Override

public void onError(final Throwable e) {

}

});

1

@FMuntenescu

Observable<Burger> burgerObservable = Observable.from(burgers);

burgerObservable.subscribe(new Subscriber<Burger>() {

@Override

public void onNext(final Burger burger) {

// handle burger 2

}@Override

public void onCompleted() {

}

@Override

public void onError(final Throwable e) {

}

});

1 2

@FMuntenescu

Observable<Burger> burgerObservable = Observable.from(burgers);

burgerObservable.subscribe(new Subscriber<Burger>() {

@Override

public void onNext(final Burger burger) {

}

@Override

public void onCompleted() {

// no more burgers!

}@Override

public void onError(final Throwable e) {

}

});

1 2

@FMuntenescu

Observable<List<Burger>>

List<Burger> burgers = new ArrayList<>();

burgers.add(new Burger());

burgers.add(new Burger());

burgerObservable =

Observable.just(burgers);

@FMuntenescu

1

Observable<List<Burger>> burgerObservable

=

Observable.just(burgers);

List<Burger> burgers = new ArrayList<>();

burgers.add(new Burger());

burgers.add(new Burger());

@FMuntenescu

Observable<List<Burger>> burgerObservable =

Observable.just(burgers);

burgerObservable.subscribe(

new Subscriber<List<Burger>>() {

@Override

public void onNext(final List<Burger> burgers) {

}

@Override

public void onCompleted() {

}

@Override

public void onError(final Throwable e) {

}

});

@FMuntenescu

Observable<List<Burger>> burgerObservable =

Observable.just(burgers);

burgerObservable.subscribe(

new Subscriber<List<Burger>>() {

@Override

public void onNext(

final List<Burger> burgers) {

// handle the list of burgers

} @Override

public void onCompleted() {

}

@Override

public void onError(final Throwable e) {

}

});

1

@FMuntenescu

Observable<List<Burger>> burgerObservable =

Observable.just(burgers);

burgerObservable.subscribe(

new Subscriber<List<Burger>>() {

@Override

public void onNext(

final List<Burger> burgers) {

// handle the list of burgers

}

@Override

public void onCompleted() {

// no more burgers!

}@Override

public void onError(final Throwable e) {

}

});

1

@FMuntenescu

Creating

Observables

From

Click

Events

@FMuntenescu

@FMuntenescu

@FMuntenescu

@FMuntenescu

[click!] [click!] [click!]

@FMuntenescu

???.subscribe(new Subscriber<String>() {

@Override

public void onNext(final String s) {

// called on every click

}

@Override

public void onCompleted() {

}

@Override

public void onError(final Throwable e) {

}

});

@FMuntenescu

[click!] [click!] [click!]

???.onNext(…) ???.onNext(…) ???.onNext(…)

@FMuntenescu

burgerButton.setOnClickListener(new View.OnClickListener() {

@Override

public void onClick(final View view) {

}

});

PublishSubject<String> burgerRequestedSubject =

PublishSubject.create();

@FMuntenescu

burgerButton.setOnClickListener(new View.OnClickListener() {

@Override

public void onClick(final View view) {

burgerRequestedSubject.onNext(…);}

});

PublishSubject<String> burgerRequestedSubject =

PublishSubject.create();

@FMuntenescu

burgerRequestedSubject.subscribe(new Subscriber<String>() {

@Override

public void onNext(final String s) {

// called on every click

}

@Override

public void onCompleted() {

// will never be called for click events

}

@Override

public void onError(final Throwable e) {

}

});

@FMuntenescu

Creating

Observables

From

Declarativ

e Code

@FMuntenescu

public Burger cook(){

// cooking is time consuming

return burger;

}

Observable<Burger> burgerObservable = Observable.fromCallable(this::cook);

@FMuntenescu

Error

Handling

@FMuntenescu

Thread t = new Thread(new Runnable() {

@Override

public void run() {

cook();

}

});

@FMuntenescu

Thread t = new Thread(new Runnable() {

@Override

public void run() {

try{

cook();

} catch(Exception e) {

// handle exception

}

}

});

@FMuntenescu

getMeatObservable()

.filter(meat -> isMeatFresh(meat))

.map(meat -> cook(meat))

.subscribeOn(

meat -> doSomething(),

error -> complain(error));

@FMuntenescu

getMeatObservable()

.filter(meat -> isMeatFresh(meat))

.map(meat -> cook(meat))

.subscribeOn(

meat -> doSomething(),

error -> complain(error));

@FMuntenescu

public Observable<Meat> getCookedMeat() {

return getMeatObservable()

.filter(meat -> isMeatFresh(meat))

.map(meat -> cook(meat));

}

Observable.zip(getBun(),

getCookedMeat(),

getTomatoSlice(),

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

.subscribeOn(

burger -> eatBurger(burger),

error -> complain(error));

@FMuntenescu

public Observable<Meat> getCookedMeat() {

return getMeatObservable()

.filter(meat -> isMeatFresh(meat))

.map(meat -> cook(meat));}

Observable.zip(getBun(),

getCookedMeat(),

getTomatoSlice(),

(bun, meat, tomato) ->

makeBurger(bun, meat, tomato))

.subscribeOn(

burger -> eatBurger(burger),

error -> complain(error));

@FMuntenescu

Start using

RxJava

@FMuntenescu

public Burger cook(){

// cooking is time consuming

// make sure it’s on a background thread

return burger;

}

Observable<Burger> burgerObservable =

Observable.fromCallable(this::cook)

.subscribeOn(Schedulers.computation())

.observeOn(AndroidSchedulers.mainThread())

@FMuntenescu

@GET(“repos”)

Call<List<Burger>> repoBurgers();

@GET(“repos”)

Observable<List<Burger>> repoBurgers();

@FMuntenescu

@FMuntenescu@FMuntenescu

jobs@updudes.net

Building a Reactive Mindset

https://upday.github.io/blog/reactive_mindset_burgers

Reactive Burgers – Code Example

https://github.com/florina-muntenescu/ReactiveBurgers

Building a BurgerReactive Mindset