Upgrade to Pro — share decks privately, control downloads, hide ads and more …

RxJava for Android

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)

Ubiratan Soares

May 07, 2016
Tweet

More Decks by Ubiratan Soares

Other Decks in Programming

Transcript

  1. 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
  2. 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 ???”
  3. 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
  4. 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 !!!
  5. 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)
  6. DADOS E SINCRONICIDADE Um valor Múltiplos valores Síncrona T getData(

    ) Iterable<T> getData( ) Assíncrona Future<T> getData( ) Observable<T> getData( )
  7. 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
  8. 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
  9. 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
  10. 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);
  11. 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);
  12. 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
  13. 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 !!!!
  14. 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
  15. 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) { … } } ); ?????????
  16. 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 ?
  17. 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
  18. 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( ) …
  19. MARBLE DIAGRAMS A B C 1 2 3 4 5

    x I onError( ) onCompleted( )
  20. MAP 1 2 3 4 5 A B C D

    E MAP { INT X -> CHAR Y }
  21. MERGE 1 3 5 2 4 1 2 3 4

    5 MERGE I I I
  22. CONCAT 1 3 5 2 4 1 3 5 2

    4 CONCAT I I I
  23. 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
  24. 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);
  25. 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);
  26. 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( ) )
  27. 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
  28. 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( )
  29. 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. "
  30. 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");
  31. 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( )
  32. 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?
  33. public class People {
 
 public String name;
 public String

    gender; public List<String> films; } public class PeopleResults {
 public List<People> results; }
  34. 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); }
  35. 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>
  36. 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'}
  37. 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
  38. 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)
  39. 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
  40. 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
  41. 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(…);
  42. 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
  43. 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
  44. 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
  45. 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);
  46. 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
  47. 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
  48. 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
  49. 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
  50. 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
  51. 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
  52. 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
  53. 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
  54. RX-LIFECYCLE public class ActivityWithObsersable extends RxActivity { @Override public void

    onResume() { super.onResume(); someObservable .compose(bindToLifecycle()) .subscribe(); } }
  55. 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;
 }
  56. 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
  57. 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
  58. 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
  59. 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
  60. 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
  61. 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