Reactive Programming with RxJava

Transcript

2. Reactive Programming with RxJava
3. RxJava I’m Hugo Cordier CTO at Melusyn ◦ Java developer ◦ Groovy developer ◦ Javascript developer
4. “Melusyn helps film makers to manage and organize their productions” We are hiring! Drop me a line at hugo@melusyn.com
5. You should download the examples I’ll use! github.com/Melusyn/ReactiveX-examples
6. RxJava ◦ Callbacks and Futures ◦ Where is Rx from? ◦ Introducing Observable ◦ Subscribing to an Observable ◦ Applying operators ◦ Handling errors ◦ Hot and Cold Observable ◦ Backpressure ◦ Schedulers
7. Callbacks and Futures
8. Rx - Callbacks server.listen(8080, "127.0.0.1", new AsyncResultHandler<Void>() { public void handle(AsyncResult<HttpServer> asyncResult) { log.info("Listen succeeded?"+asyncResult.succeeded()); } } ); Lot of boilerplate code Hard to read A mess to chain
9. Rx - CompletableFutures CompletableFuture<Customer> customerFuture = loadCustomerDetails(123); CompletableFuture<Shop> shopFuture = closestShop(); CompletableFuture<Route> routeFuture = customerFuture.thenCombine(shopFuture, (cust, shop) -> findRoute(cust, shop) ); Easier to chain Easier to read But...
10. Futures contains only 1 value We can’t deal about time It’s hard to deal with threads
11. Where is Rx from?
12. Rx - Where’s it from ◦ Erik Meijer developed the concepts at Microsoft ◦ Microsoft open sourced Rx.NET and RxJS in 2012 ◦ In 2013 Netflix open sourced RxJava, based on Erik Meijer concepts
13. Introducing Observable
14. Observable - Iterable parallel Single return value Mutiple return values Synchronous Data getData() Iterable<Data> getData() Asynchronous Future<Data> getData() Observable<Data> getData()
15. Observable - Iterable parallel Data data = getData() if (data.equals(x)) { // Do something } else { // Do something else } Single return value Mutiple return values Synchronous Data getData() Iterable<Data> getData() Asynchronous Future<Data> getData() Observable<Data> getData()
16. Observable - Iterable parallel Iterable<Data> data = getData() data.forEach(value -> { if (value.equals(x)) { // Do something } else { // Do something else } }) Single return value Mutiple return values Synchronous Data getData() Iterable<Data> getData() Asynchronous Future<Data> getData() Observable<Data> getData()
17. Observable - Iterable parallel CompletableFuture<Data> data = getData() data.thenApply(value -> { if (value.equals(x)) { // Do something } else { // Do something else } }) Single return value Mutiple return values Synchronous Data getData() Iterable<Data> getData() Asynchronous Future<Data> getData() Observable<Data> getData()
18. Observable - Iterable parallel Observable<Data> data = getData() data.map(value -> { if (value.equals(x)) { // Do something } else { // Do something else } }) Single return value Mutiple return values Synchronous Data getData() Iterable<Data> getData() Asynchronous Future<Data> getData() Observable<Data> getData()
19. An Observable can : • emit events (0 to ∞) • emit an error • complete stream Once an error has been emitted, no further events can be. An Observable could be infinite. Observable - Introduction
20. rx.Observable.create { observer -> try{ for (int i = 0; i < 2; i++) { observer.onNext(i); } observer.onCompleted(); } catch (Exception e) { observer.onError(e); } }
21. rx.Observable.create { observer -> executor.execute(new Runnable() { def void run() { try { for (int i = 0; i < 10; i++) { observer.onNext(i); this.sleep(2000) } observer.onCompleted(); } catch (Exception e) { observer.onError(e); } } }) }
22. Subscribing to an Observable Consuming data
23. getData() .subscribe(new Subscriber() { @Override void onNext(Object o) { logger.info(o) } @Override void onError(Throwable e) { logger.error("Oops", e) } @Override void onCompleted() { logger.debug("Stream has completed") } })
24. getData() .subscribe( { next -> logger.info(o) }, { e -> logger.error("Oops", e)}, { logger.debug("Stream has completed")} )
25. Live examples https://github.com/Melusyn/ReactiveX-examples
26. Applying operators Transform the data
27. This is a Marble Diagram Live examples : http://rxmarbles.com/#filter getData().filter {it >= 0}
28. rx.Observable.merge(getDataA(), getDataB()) .subscribe {value -> logger.info value} Live example : http://rxmarbles.com/#merge
29. rx.Observable.zip(getDataA(), getDataB(), {vA, vB -> [vA, vB] }) .subscribe {value -> logger.info value} Live example : http://rxmarbles.com/#zip
30. Observable<String> getDataA(Integer param) {...} rx.Observable.range(1, 10) .flatMap(i -> getDataA(i)) .subscribe {String s -> logger.info s}
31. Live examples https://github.com/Melusyn/ReactiveX-examples
32. Let’s use swapi.co, the Star Wars API Films, Characters, Planets, Species, everything from Star Wars :D What if I want to fetch every Character of a film, consolidated with its homeworld planet and species? Let’s try with Rx! Real life exemple
33. Handling errors Oops! What should I do?
34. getData() .filter {it>3} .map {it+2} .subscribe( { next -> logger.info(o) }, { e -> logger.error("Oops", e) // Handle any error here }, { logger.debug("Stream has completed")} ) Once Observable has failed, it terminates and do not send any more events.
35. Errors can be caught with Observable<T> onErrorResumeNext (Observable<? extends T> resumeSequence) Observable<T> onErrorReturn (Func1<java.lang.Throwable,? extends T> resumeFunction)
38. Live examples https://github.com/Melusyn/ReactiveX-examples
39. Hot and Cold Observable behaviour
40. Cold Observables They are passive streams, producing events when subscribed to. This is default behaviour. def source = rx.Observable .interval(1, java.util.concurrent.TimeUnit. SECONDS) .doOnEach {println "Emitted an event"} source.subscribe ( {println it}, {it.printStackTrace()} ) This part is declarative. No events are emitted until something subscribe to it At this point, source starts to emit events
41. Hot Observables They are active streams, and produce events regardless of subscription def source = rx.Observable .interval(1, java.util.concurrent.TimeUnit.SECONDS) .doOnEach {println "Emitted an event"} .publish() source.connect() source.subscribe ( {println it}, {it.printStackTrace()} ) publish() makes it a hot observable which will be triggered by connect() At this point, source starts to emit events
42. Live examples https://github.com/Melusyn/ReactiveX-examples
43. Backpressure Controlling a stream from its subscriber
44. Backpressure As Hot Observables emit events as their own pace, what if it emits faster than subscriber can consume data? Backpressure allows subscriber to control the stream. Some existing operators implements backpressure : merge, zip, from, groupBy, delay, scan, ... It can be implemented in any Observable source.
45. Live examples https://github.com/Melusyn/ReactiveX-examples
46. Schedulers Where to run the source and operators?
47. Schedulers Rx is single threaded by default Schedulers define where to run each computation Each Operator has a default Scheduler [1] Rx allows developers to specify where to run their code using : public final Observable<T> subscribeOn(Scheduler scheduler) public final Observable<T> observeOn(Scheduler scheduler) [1] And you can find them here
48. Schedulers Scheduler Purpose Schedulers.computation(  ) meant for computational work such as event- loops and callback processing; do not use this scheduler for I/O (use Schedulers.io(  ) instead) Schedulers.from(executor) uses the specified Executor as a Scheduler Schedulers.immediate(  ) schedules work to begin immediately in the current thread Schedulers.io(  ) meant for I/O-bound work such as asynchronous performance of blocking I/O, this scheduler is backed by a thread-pool that will grow as needed; for ordinary computational work, switch to Schedulers.computation(  ) Schedulers.newThread(  ) creates a new thread for each unit of work Schedulers.trampoline(  ) queues work to begin on the current thread after any already-queued work
49. Live examples https://github.com/Melusyn/ReactiveX-examples
50. Going further Reactive programming world
51. Rx World • Rx.NET • RxJava, RxScala, RxGroovy, RxJRuby, RxNetty, RxAndroid • Rx.rb • Rx.py • RxClojure • RxCpp • RxKotlin
52. Thanks! ANY QUESTIONS? You can find me at @HugoCrd hugo@melusyn.com