RxJava for Android

Versão completa dos meus slides de apoio para a apresentação sobre RxJava : fundamentos, aplicações para Android, comparação com Java8 Streams e mais.

Apresentação realizada nos seguintes eventos :

- QConSP 2016 (abril/2016)
- BEV (LuizaLabs) (maio/2016)
- Android Meetup de 3 anos (maio/2016)
- JavaOne Latin America (junho/2016)
- The Developers Conference (julho/2016)
- RxJava Workshop by OpenSanca (julho/2016)
- Androidos Day (julho/2016)
- GDG Campinas Devfest (novembro/2016)

1. RXJAVA PROGRAMAÇÃO REATIVA FUNCIONAL UBIRATAN SOARES JULHO / 2016
2. MOTIVAÇÃO Java8 trouxe uma API de Streams, que permite operações comuns sobre coleções como Map, Filter, Reduce e Collect Java9 trará uma API de ReactiveStreams, adotadando muitos conceitos presentes em RxJava diretamente no JDK Funcional Programming ajuda a escrever código mais robusto e conciso de maneira geral
3. ERIK MEIJER “Do ponto de vista de fluxo de dados, qual a diferença entre consultar o DB e processar as posições do ponteiro do mouse ???”
4. VELHOS PROBLEMAS SEMPRE NA MODA Execução orquestrada de código de forma assíncrona Execução e sincronização de processamento concorrente Tratamento de erros Escalabilidade
5. UMA NOVA FORMA DE PENSAR E se ao invés de obter dados de forma síncrona, os dados chegassem até mim de forma assíncrona ? Reativo : algo que reage a um estímulo !!! Fundamento matemático : teoria da categorias !!!
6. REACTIVE MANIFESTO As demandas atuais tipicamente pedem aplicações Responsivas Resilientes Orientadas a eventos Escaláveis
7. HANS DOCKTER “Programar é arte de encontrar as abstrações corretas”
8. RELEITURA DE CONCEITOS Pensamento em termos de fluxo de dados : eventos discretos e fluxo desses eventos É possível reagir a eventos e combinar os mesmos O estado do sistema deve mudar conforme a passagem de eventos no tempo Eventos no fluxo são imutáveis Sistema que idealmente nunca bloqueia (I/O, cálculos, etc)
9. DADOS E SINCRONICIDADE Um valor Múltiplos valores Síncrona T getData( ) Iterable<T> getData( ) Assíncrona Future<T> getData( ) Observable<T> getData( )
10. RELEITURA DE CONTRATOS Iterable<T> Observer<T> Obter o próximo T next( ) onNext( T ) Sinalizar erro throws Exception( ) onError( Thowable ) Saber se terminou hasNext( ) onComplete( ) PULLED WAY PUSHED WAY
11. CONCEITOS BÁSICOS OBSERVABLE OBSERVER OPERATOR SCHEDULER
12. OBSERVABLE Representa o fluxo de dados (ou eventos, ou itens emitidos) Por padrão, executa de forma sequencial (não concorrente) Repassa cada item emitido para um observador (callback) Pode ser associado ao conceito de source da Streams API do java8, porém seguindo push model quanto aos dados
13. CRIANDO OBSERVABLES (I) Observable<String> source = Observable.just("GOOGLE", "APPLE", "MICROSOFT"); source.subscribe(new Action1() { @Override public void call(String s) { System.out.println(s) } }); GOOGLE APPLE MICROSOFT PROCESS FINISHED WITH EXIT CODE 0
14. CRIANDO OBSERVABLES (II) Observable<String> source = Observable.just("GOOGLE", "APPLE", “MICROSOFT"); source.subscribe((String)company -> System.out.println(company)); Observable.just("GOOGLE", "APPLE", "MICROSOFT") .subscribe(System.out::println);
15. CRIANDO OBSERVABLES (III) Observable.fromCallable(() -> “RxJava is Awesome”) .subscribe(System.out::println); List<String> names = Arrays.asList("Banana", "Apple", "Orange"); Observable.from(names).subscribe(System.out::println);
16. CRIANDO OBSERVABLES (IV) Observable.create( new Observable.OnSubscribe<SomeEventType>() { @Override public void call( final Subscriber<? super SomeEventType> subscriber) { . . . // Wrap some callback and emits new events } }) ); BEW ARE, THINK TW ICE
17. CRIANDO OBSERVABLES (V) Observable.interval(1, TimeUnit.SECONDS) .map(time -> "AT SECOND " + time) .subscribe(System.out::println); AT SECOND 0 AT SECOND 1 AT SECOND 2 AT SECOND 3 … WARNING : estamos roubando aqui, se você executar esse código, não verá esse log !!!!
18. OBSERVER / SUBSCRIBER Consome o fluxo de dados Respeita o contrato no estilo pushed data É o ponto no qual os erros flutuam : um erro que aconteça durante uma operação interrompe a sequência de emissões Callbacks sinalizam o término da sequência Adições ao Observer Pattern do GOF
19. OBSERVER public interface Observer<T> { } void onCompleted(); void onError(Throwable e); void onNext(T data);
20. source.subscribe(new Observer<Item>() { @Override public void onCompleted() { … } @Override public void onError(Throwable e) { … } @Override public void onNext(Item item) { … } } ); source.subscribe(new Subscriber<Item>() { @Override public void onCompleted() { … } @Override public void onError(Throwable e) { … } @Override public void onNext(Item item) { … } } ); ?????????
22. Observable.fromCallable(() -> “Better subscribing with actions”) .subscribe( System.out::println, throwable -> someErrorAction(), () -> done() ); COM ACTIONS
23. PIPELINE DE OPERAÇÕES Observable.from(…) .flatMap(…) .filter(…) .map(…) .observeOn(…) .subscribeOn(…) .subscribe(…); upstream sequence downstream sequence
24. Observable.zip( restApi.getAvaliableItems(), restApi.getRecommendedItems(clientId), new ItemsResultsZipper()) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Observer<List<Item>>() { @Override public void onCompleted() { … } @Override public void onError(Throwable e) { … } @Override public void onNext(List<Item> items) { … } }); podemos fazer melhor ?
25. OPERADORES Funções que permitem manipular a sequência de eventos observáveis, sejam os itens emitidos, seja a própria sequência (ou múltiplas delas) Reactive Extensions define um enorme catálogo de operadores quanto à semântica, em categorias bem definidas
26. CATEGORIA OPERADORES CRIAÇÃO just( ), from( ), range( ), interval( ), defer( ) … COMBINAÇÃO zip( ), merge( ), combineLatest( ), concat( ) … TRANSFORMAÇÃO map( ), flatMap( ), concatMap( ) … FILTRAGEM filter( ), take( ), skip( ), debounce( ) … MUITO MAIS ! cache( ), replay( ), retry( ), retryWhen( ) …
27. MARBLE DIAGRAMS A B C D E onNext( ) é chamado cinco vezes
28. MARBLE DIAGRAMS A B C 1 2 3 4 5 x I onError( ) onCompleted( )
29. MAP 1 2 3 4 5 A B C D E MAP { INT X -> CHAR Y }
30. FILTER 1 2 3 4 5 2 4 FILTER { INT X -> INT Y}
31. MERGE 1 3 5 2 4 1 2 3 4 5 MERGE I I I
32. CONCAT 1 3 5 2 4 1 3 5 2 4 CONCAT I I I
33. AMB 1 3 5 2 4 1 3 5 AMB I I I
34. INTERVAL INTERVAL {} 1 2 3 4 5
35. rxmarbles.com Disponível como app Android
36. SCHEDULER Escalonador de trabalho entre thread pools distintas Abstração em volta de Executors ( java.util.concurrent ) Qualquer Observable pode ser produzido em um Scheduler e observado em outro Mecanismo fundamental para aplicações móveis
37. Observable.range(1, 2) .map(i -> i * 2) .observeOn(Schedulers.io()) .doOnNext(i -> System.out.println( "Emitting " + i + " on thread " + threadName())) .observeOn(Schedulers.computation()) .map(i -> i * 10) .subscribe(i -> System.out.println( "Received " + i + " on thread " + threadName())); Emitting 2 on thread RxCachedThreadScheduler-1 Received 20 on thread RxComputationThreadPool-3 Emitting 4 on thread RxCachedThreadScheduler-1 Received 40 on thread RxComputationThreadPool-3 sleep(3000);
38. Observable.just("Google", "Apple", "Microsoft", "IBM") .subscribleOn(Schedulers.computation()) .subscribe(s -> System.out.println( "Received " + s + " on thread " + threadName())); Received Google on thread RxComputationThreadPool-1 Received Apple on thread RxComputationThreadPool-1 Received Microsoft on thread RxComputationThreadPool-1 Received IBM on thread RxComputationThreadPool-1 sleep(3000);
39. NUÂNCIAS SOBRE SCHEDULERS (I) Schedulers.io( ) encapsula um thread pool de tamanho variável Schedulers.computation( ) encapsula um thread pool de tamanho fixo Operam sobre deamon threads São extensíveis para encapsular threads importantes, como por exemplo a UI thread do Android ou do JavaFX Alguns factory methods de Observable já associam um scheduler à sequência (como interval( ) )
40. NUÂNCIAS SOBRE SCHEDULERS (II) subscribeOn( ) instrui o framework sobre em qual thread os itens as emissões iniciam para o consumidor observeOn( ) instrui o framework sobre quais schedulers podem operar nas etapas intermediárias do pipeline de operações subscribeOn( ) em geral é usado uma vez no, início do pipeline observeOn( ) é utilizado conforme a necesside de concorrência nas etapas do pipeline
41. TRATAMENTO DE ERROS Evento terminal destrói a sequência. Ponto. Recuperação via operadores onErrorResumeNext( ), onErrorReturn( ) e outros Sugestão : implemente um wrapper com o retorno de onError( throwable), isso ajudará no stacktrace em casos de erro Sugestão : verificar semântica de onExceptionResumeNext( )
42. CONCEITOS AVANÇADOS
43. FLATMAP "Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable The FlatMap operator transforms an Observable by applying a function that you specify to each item emitted by the source Observable, where that function returns an Observable that itself emits items. FlatMap then merges the emissions of these resulting Observables, emitting these merged results as its own sequence. This method is useful, for example, when you have an Observable that emits a series of items that themselves have Observable members or are in other ways transformable into Observables, so that you can create a new Observable that emits the complete collection of items emitted by the sub-Observables of these items. "
45. FLATTENING (MERGE) MAPPING (PROVIDED FUNCTION)
46. List<String> companies = Arrays.asList("Google", "Apple", "Microsoft"); Google Apple Microsoft Larry Steve Bill flatMap( ) desmontou 2 sequências de items e juntou individualmente os itens de cada sequência em uma única sequência final mapeamento Observable<List<String>> obs = Observable.just(companies, leaders); Observable<String> flat = obs.flatMap(strings -> Observable.from(strings)); flat.subscribe(System.out::println); List<String> leaders = Arrays.asList("Larry", "Steve", "Bill");
47. MAIS SOBRE FLATMAP flatmap( ) transforma em sequência em outra através da função de mapeamento flatmap( ) é um dos mecanismos para encadeamento de operações assíncronas flatmap( ) não garante a ordem dos itens emitidos na sequência gerada final após o merging; se a ordem dos itens for importante, usar concatMap( )
48. ESTUDO DE CASO : STAR WARS REST API https://swapi.co/ Consumir com Retrofit2 Consulta simples em um endpoint : lista de personagens Consulta encadeada em 2 endpoints : dado um personagem aleatório, em quais filmes ele aparece?
50. public class People {
 
 public String name;
 public String gender; public List<String> films; } public class PeopleResults {
 public List<People> results; }
51. public class Movie {
 
 public String title;
 public String director; public String releaseDate; }
52. public interface StarWarsAPI { @GET(“people“) Observable<PeopleResults> people(); @GET(“film/{film_id}“) Observable<Movie> movieById( @Path("film_id") String id, ); } public StarWarsAPI createAPI() {
 
 Retrofit retrofit = new Retrofit.Builder()
 .baseUrl(API_URL)
 .addConverterFactory(GsonConverterFactory.create())
 .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
 .build();
 
 return retrofit.create(StarWarsAPI.class); }
53. starWarsAPI.people() .subscribeOn(Schedulers.io()) .flatMap(payload -> Observable.from(payload.results)) .subscribe(
 System.out::println,
 Throwable::printStackTrace,
 () -> System.out::println("Done -> All People")
 ); PEOPLE{NAME='LUKE SKYWALKER', GENDER='MALE'} PEOPLE{NAME='C-3PO', GENDER='N/A'} PEOPLE{NAME='R2-D2', GENDER='N/A'} PEOPLE{NAME='DARTH VADER', GENDER=‘MALE'} [ … ] PEOPLE{NAME='OBI-WAN KENOBI', GENDER='MALE'} DONE -> ALL PEOPLE // List<Person>
54. starWarsApi.people() .subscribeOn(Schedulers.io()) .flatMap(payload -> selectRandomPeople(payload.results)) .doOnNext(System.out::println) .flatMap(people -> Observable.from(people.films)) .flatMap(filmUrl -> { String filmId = ResourceIdExtractor.idFromUrl(filmUrl); return api.movieById(filmId) .subscribeOn(Schedulers.io());
 }) .subscribe(System.out::println); PEOPLE{NAME='OWEN LARS', GENDER=‘MALE'} MOVIE{TITLE='ATTACK OF THE CLONES', EPISODE=2, DIRECTOR='GEORGE LUCAS', RELEASEDAY='2002-05-16'} MOVIE{TITLE='A NEW HOPE', EPISODE=4, DIRECTOR='GEORGE LUCAS', RELEASEDAY='1977-05-25'} MOVIE{TITLE='REVENGE OF THE SITH', EPISODE=3, DIRECTOR='GEORGE LUCAS', RELEASEDAY='2005-05-19'}
55. OBSERVABLES ESPECIAIS (I) Observable.empty( ) é uma sequêcia que finaliza sem emitir itens, com sucesso Observable.error( ) aceita um Throwable e emite uma sequência com o estado terminal de erro Observable.never( ) é uma sequência que não emite nenhum item, nunca Observable.toBlocking( ) é um mecanismo para executar a emissão de itens de modo síncrono
56. Single é um Observable que emite um único item Completable é uma sequência que não emite nenhum item, mas emite os eventos terminais Retrofit2 pode proxyar um Single (2.0.0 final em diante) Completable é um Observable (muito!) mais leve para interessados no sucesso ou fracasso de uma sequência OBSERVABLES ESPECIAIS (II)
57. Single<Integer> single = Observable.range(1, 10).first().toSingle(); single.subscribe(System.out::println); // 1 Observable source = Observable.just("Lets See"); Completable.fromObservable(source)
 .subscribe(() -> System.out.println(“YES!")); // YES Observable inevitable = Observable.error(new Exception("Ouch!")); Completable.fromObservable(inevitable)
 .subscribe(
 throwable -> System.out.println("FAIL"),
 () -> System.out.println("DONE”) ); // FAIL
58. TRANSFORMERS Aplicáveis através do operador compose( ) Manipulam a sequência como um todo, ideais para evitar repetições de código em etapas comuns do pipeline Casos comuns : setup de schedulers, composição de filtros, etc
59. restApi.callMethod01() .subscribleOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(…); restApi.callMethod02() .subscribleOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(…); DRY ?
60. public static <T> Observable.Transformer<T, T> setupSchedulers() { return observable -> observable.subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } restApi.callMethod01() .compose(Transformers.setupSchedulers() ) .subscribe(…); restApi.callMethod02() .compose(Transformers.setupSchedulers() ) .subscribe(…);
61. SUBSCRIPTION Subscription representa o resultado de um Observer / Subscriber ser associado a um Observable Permitem deixar de acompanhar a sequência “por fora”, eg, de forma explícita via unsubscribe( ) Mecanismo possível para evitar memory leaks no Android Múltiplas sequências podem ser agrupadas via CompositeSubscription
62. Subscription first = Observable.interval(1, TimeUnit.SECONDS)
 .subscribe(System.out::print);
 
 Subscription second = Observable.range(1, 100000)
 .subscribe(System.out::print);
 
 CompositeSubscription subs = new CompositeSubscription();
 subs.add(first);
 subs.add(second); // . . . 
 first.unsubscribe(); subs.add(third); // . . . 
 if(subs.hasSubscriptions()) subs.unsubscribe(); Desinscreve-se de todos as sequências participantes
63. SUBJECTS Subject é alguém que implementa tanto Observable quanto Observer Também aceitam múltiplos Subscribers Quatro variantes : Async, Behavior, Publish e Replay Ideais para particionar uma sequência em vários fluxos de eventos diferentes Bastante flexíveis, porém traiçoeiros
64. PUBLISH SUBJECT SUBSCRIBE SUBSCRIBE
65. PublishSubject<Integer> subject = PublishSubject.create(); From first : 1 From second : ODD From first : 2 From second : EVEN subject.subscribe(
 integer -> System.out.println("From first : " + integer),
 throwable -> reportError(),
 () -> done()
 ); subject.map(number -> number / 2 != 0 ? "ODD" : "EVEN")
 .subscribe(s -> System.out.println("From second : " + s)); Observable.range(1, 2).subscribe(subject);
66. HOT AND COLD OBSERVABLES Hot Observables começam a emitir itens no momento em que são criados Cold Observables começam a emitir itens no momento em que são observados Hot Observables são passíveis de perda de emissões Cold Observables podem produzir side-effects Alguns operadores transformam a temperatura da sequência
67. BACKPRESSURE Cenário no qual a sequência observável (produtor) emite mais itens do que o observador (consumidor) consegue processar Menos provável em sequências de itens de natureza estritamente computacional Mais provável em sequências sendo geradas e consumidas por IO
68. LIDANDO COM BACKPRESSURE Dois tipos de cenário, de acordo com a natureza dos itens emitidos Perda de emissões implica em perda de informação Perda de emissões não implica perda de informação
69. BACKPRESSURE COM PERDA DE DADOS Estratégia : diminuir a quantidade de dados que chegam até o consumidor segundo alguma heurística de perda Diversos operadores disponíveis : sample( ), debounce( ), take( ), throttleFirst( ), etc
70. BACKPRESSURE SEM PERDA DE DADOS Duas estratégias imediatas (i) Bufferizar dados para processamento posterior, com operadores como cache( ), buffer( ), window( ) etc (ii) Informar a upstream sequence que é preciso diminuir o ritmo na emissão de itens, usando Producer API
71. JAVA8 STREAMS List<String> partners = Arrays.asList(
 "Caelum", “Google", "7Comm", "Porto", "Oxigenio",
 "Intel", "IBM", "Globalcode", "Samsung", "Novatec",
 "Paypal", "Twitter", "Facebook", "USP", "Impacta" );
 
 Optional<Integer> result = partners.parallelStream()
 .filter(s -> s.length() > 3)
 .map(String::length)
 .distinct()
 .reduce((a, b) -> a + b);
 
 if(result.isPresent()) System.out.println(result.get()); // 36
72. RXJAVA VS JAVA8 STREAMS (I) Streams são desenhadas para processamento computacional, in-memory e com fácil paralelização Streams trazem Collectors API para melhor semântica de agrupamentos Streams pode adaptar sources através de Spliterators API Streams operam estritamente em pull mode, eg, não baseado em callbacks Não há backpressure
73. RXJAVA VS JAVA8 STREAMS (II) RxJava opera essencialmente em push mode, eg, baseado em callbacks RxJava contém a noção de tempo na API, permitindo inclusive sincronização de eventos no tempo RxJava oferece controle granular sobre concorrência via Schedulers RxJava é desenhada tanto para uso de processamento computacional quanto para IO (onde brilha muito!) RxJava tende a ser mais burocrática e/ou pesada que Streams para processamento computacional tipo map/filter/collect/reduce
74. E NO DIA A DIA ?? Mas afinal …
75. RX-LIFECYCLE public class ActivityWithObsersable extends RxActivity { @Override public void onResume() { super.onResume(); someObservable .compose(bindToLifecycle()) .subscribe(); } }
76. RX-BINDING @Override public boolean onCreateOptionsMenu(Menu menu) { getMenuInflater().inflate(R.menu.menu_search_by_terms, menu);
 MenuItem search = menu.findItem(R.id.search);
 SearchView searchView = (SearchView) MenuItemCompat.getActionView(search);
 
 RxSearchView.queryTextChangeEvents(searchView)
 .throttleLast(100, TimeUnit.MILLISECONDS)
 .debounce(300, TimeUnit.MILLISECONDS)
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe(this::proceedWithQuery);
 
 return true;
 }
77. ¯\_(ϑ)_/¯
79. PROBLEMAS COM RXJAVA Debugging pode ser difícil (stacktraces gigantescos) Verbosidade (Java 6,7) Uso descuidado pode levar a problema de leaks memória Curva de aprendizado : muitos conceitos, em geral complexos, que exigem atenção aos detalhes
80. CONCLUSÕES RxJava é (muito!) superior do que opções do framework do Android para concorrência (eg AsyncTask) RxJava permite execução orquestrada de código assíncrono e/ ou concorrente de forma declarativa RxJava permite operacionais funcionais sobre dados de maneira retrocompatível RxJava é BATTLE-TESTED RxJava ajuda a resolver melhor problemas difíceis
81. REFERÊNCIAS (I) "Functional Reactive Programming with RxJava" by Ben Christensen https://youtu.be/_t06LRX0DV0 “Learning RxJava (for Android) by example“ by Kaushik Goupal https://youtu.be/k3D0cWyNno4 “Demystifying RxJava Subscribers" by Jake Wharton https://youtu.be/NVKmyK6sd-Q “What does it mean to be Reactive ?” by Erik Meijer https://youtu.be/sTSQlYX5DU0
82. REFERÊNCIAS (II) "Grokking RxJava Series” by Dan Lew http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/ “The Introduction to Reactive Programming you`ve been missing” by André Staltz https://gist.github.com/staltz/868e7e9bc2a7b8c1f754 Oficial RxJava Wiki by NetFlix https://github.com/ReactiveX/RxJava/wiki Advanced RxJava Blog by David Karnok akarnokd.blogspot.com
83. REFERÊNCIAS (III) GradleLambda : https://github.com/evant/gradle-retrolambda RxAndroid : https://github.com/ReactiveX/RxAndroid RxLifecycle : https://github.com/trello/RxLifecycle RxBinding : https://github.com/JakeWharton/RxBinding RxRelay : https://github.com/JakeWharton/RxRelay Frodo : https://github.com/android10/frodo
84. UBIRATAN SOARES Computer Scientist by ICMC/USP Software Engineer @ Luiza Labs Google Developer Expert for Android Intel Software Innovator for Android Teacher, speaker, etc, etc
85. OBRIGADO THAT`S ALL FOLKS !!! @ubiratanfsoares speakerdeck.com/ubiratansoares ubiratansoares.github.io