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

Architecture découplée grâce aux Reactive Exten...

David
April 20, 2016

Architecture découplée grâce aux Reactive Extensions

Auparavant, les applications se connectaient à une seule base de données. Aujourd’hui, elles doivent communiquer avec un ensemble de base de données et composer avec des services web distants. Elles ne sont plus des blocs monolithiques, mais distribuées sur différents serveurs sur différentes zones géographiques. Elles sont “scalables” mais elles doivent maintenant attendre les réponses d’autres applications.

Comment profiter de ces temps d’attente ? Comment casser ce couplage temporel ? Comment transformer un couplage applicatif fort en couplage lâche ?

Cette université fera le tour de RxJava et RxJs. Comment les mettre en place sur un projet et dans une équipe. Et surtout comment utiliser ces bibliothèques pour découpler votre architecture.

University - 20 Avril 2016 - Devoxx France (Palais des congrès de Paris)

David Wursteisen - Brice Dutheil

David

April 20, 2016
Tweet

More Decks by David

Other Decks in Programming

Transcript

  1. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ce que nous allons voir Reactive

    Extensions en théorie en pratique sur notre application
  2. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ce que nous allons voir Reactive

    Extensions en théorie en pratique sur notre application Pause
  3. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ce que nous ne verrons pas

    Énumération des opérateurs Mieux que ces alternatives ?
  4. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ce que nous ne verrons pas

    Énumération des opérateurs Mieux que ces alternatives ? Expliquer tous les patterns d’utilisation possible
  5. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête connexion établie prête à être consommée
  6. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread Accepte la requête connexion établie prête à être consommée acceptedCount =
  7. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Worker Executor maxThreads =
  8. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Thread
  9. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Thread w a r Lit la payload war
  10. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Thread w a r Lit la payload Business war
  11. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Thread w a r Lit la payload Écrit la réponse, ferme la connexion Business war
  12. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / BIO client OS Négociation

    TCP Requête # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Thread war Lit la payload Écrit la réponse, ferme la connexion Business
  13. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat client OS Négociation TCP Requête

    # nouvelle requête Acceptor Thread connexion établie prête à être consommée Accepte la requête Requête donnée aux threads de travail si slot disponible Thread w a r Lit la payload Écrit la réponse, ferme la connexion Business war
  14. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat client OS Requête # nouvelle

    requête Acceptor Thread Attend un slot disponible Thread war Executor full
  15. @BriceDutheil @dwursteisen #DevoxxFR #Rx Point de vue du CPU Pas

    d’instruction à exécuter Le système attend sur des I/Os
  16. @BriceDutheil @dwursteisen #DevoxxFR #Rx Localement aussi Lecture/Écriture sur disque -

    ex : swap Panne hardware - ex : disque en panne augmente la latence
  17. @BriceDutheil @dwursteisen #DevoxxFR #Rx NIO ne résout pas le problème

    de fond §9.2 The Container Provider should ensure that the dispatch of the request to a target servlet occurs in the same thread of the same JVM as the original request. http://download.oracle.com/otn-pub/jcp/servlet-3.0-mrel-full-oth-JSpec/servlet-3_0-mrel-spec.pdf
  18. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / NIO client OS Requête

    # nouvelle requête Acceptor Thread Thread Poller Thread Requête # war
  19. @BriceDutheil @dwursteisen #DevoxxFR #Rx Tomcat / NIO client OS Requête

    # nouvelle requête Acceptor Thread Thread Poller Thread Requête # war
  20. @BriceDutheil @dwursteisen #DevoxxFR #Rx Améliorer le code de notre application

    • Executors • Future • Embarquer une techno à la ActiveMQ ?
  21. @BriceDutheil @dwursteisen #DevoxxFR #Rx Améliorer le code de notre application

    • Executors • Future • Embarquer une techno à la ActiveMQ ? • Problème de lourdeur, overhead
  22. @BriceDutheil @dwursteisen #DevoxxFR #Rx Améliorer le code de notre application

    • Executors • Future • Embarquer une techno à la ActiveMQ ? • Problème de lourdeur, overhead • Quid de la composition
  23. @BriceDutheil @dwursteisen #DevoxxFR #Rx Améliorer le code de notre application

    • Executors • Future • Embarquer une techno à la ActiveMQ ? • Problème de lourdeur, overhead • Quid de la composition • CompletableFuture => Avec Collections ou Streams ?
  24. @BriceDutheil @dwursteisen #DevoxxFR #Rx Améliorer le code de notre application

    • Executors • Future • Embarquer une techno à la ActiveMQ ? • Problème de lourdeur, overhead • Quid de la composition • CompletableFuture => Avec Collections ou Streams ?
  25. @BriceDutheil @dwursteisen #DevoxxFR #Rx Améliorer le code de notre application

    • Executors • Future • Embarquer une techno à la ActiveMQ ? • Problème de lourdeur, overhead • Quid de la composition • CompletableFuture => Avec Collections ou Streams ? • Gestion d’erreurs
  26. @BriceDutheil @dwursteisen #DevoxxFR #Rx Temporalité http://blog.codinghorror.com/the-infinite-space-between-words/ Temps machine Temps humain

    1 cycle CPU 0.3 nanosecondes 1 seconde Accès disque (SSD) 50-150 microsecondes 2-6 jours Accès disque (mécanique) 1-10 millisecondes 1-12 mois Envoi d’un paquet de San Francisco à New York 40 milliseconds 4 ans
  27. @BriceDutheil @dwursteisen #DevoxxFR #Rx Temporalité http://blog.codinghorror.com/the-infinite-space-between-words/ Temps machine Temps humain

    1 cycle CPU 0.3 nanosecondes 1 seconde Accès disque (SSD) 50-150 microsecondes 2-6 jours Accès disque (mécanique) 1-10 millisecondes 1-12 mois Envoi d’un paquet de San Francisco à New York 40 milliseconds 4 ans
  28. @BriceDutheil @dwursteisen #DevoxxFR #Rx Pour quoi faire ? future1.get(); future2.get();

    future3.get(); future4.get(); future5.get(); Ordonnancement optimal ?
  29. @BriceDutheil @dwursteisen #DevoxxFR #Rx Pour quoi faire ? client.execute(new Callback()

    { @Override public void completed(HttpResponse response) { } });
  30. @BriceDutheil @dwursteisen #DevoxxFR #Rx Pour quoi faire ? client.execute(new Callback()

    { @Override public void completed(HttpResponse response) { client.execute(new Callback() { @Override public void completed(HttpResponse response) { } }); } });
  31. @BriceDutheil @dwursteisen #DevoxxFR #Rx Pour quoi faire ? client.execute(new Callback()

    { @Override public void completed(HttpResponse response) { client.execute(new Callback() { @Override public void completed(HttpResponse response) { client.execute(new Callback() { @Override public void completed(HttpResponse response) { } }); } });
  32. @BriceDutheil @dwursteisen #DevoxxFR #Rx Pour quoi faire ? client.execute(new Callback()

    { @Override public void completed(HttpResponse response) { client.execute(new Callback() { @Override public void completed(HttpResponse response) { client.execute(new Callback() { @Override public void completed(HttpResponse response) { } }); } }); Callback hell
  33. @BriceDutheil @dwursteisen #DevoxxFR #Rx λ ϕ φ json json json

    Observable<json> Observable<Integer> Observable<Click>
  34. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(new Observer<Integer>() {

    @Override public void onNext(Integer v) { System.out.println("Valeur : " + v); } @Override public void onCompleted() { System.out.println("Fin"); } @Override public void onError(Throwable e) { System.out.println("Erreur"); } });
  35. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(new Observer<Integer>() {

    @Override public void onNext(Integer v) { System.out.println("Valeur : " + v); } @Override public void onCompleted() { System.out.println("Fin"); } @Override public void onError(Throwable e) { System.out.println("Erreur"); } });
  36. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(new Observer<Integer>() {

    @Override public void onNext(Integer v) { System.out.println("Valeur : " + v); } @Override public void onCompleted() { System.out.println("Fin"); } @Override public void onError(Throwable e) { System.out.println("Erreur"); } });
  37. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(new Observer<Integer>() {

    @Override public void onNext(Integer v) { System.out.println("Valeur : " + v); } @Override public void onCompleted() { System.out.println("Fin"); } @Override public void onError(Throwable e) { System.out.println("Erreur"); } });
  38. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(v-> System.out.println("Valeur :

    " + v), e -> System.out.println("Erreur"), () -> System.out.println("Fin") );
  39. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(v-> System.out.println("Valeur :

    " + v), e -> System.out.println("Erreur"), () -> System.out.println("Fin") ); onNext onError onCompleted
  40. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .map(i -> i

    * 2) .take(2) .defaultIfEmpty(0) .subscribe(); Opérateur map
  41. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .map(i -> i

    * 2) .take(2) .defaultIfEmpty(0) .subscribe(); Opérateur take
  42. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .map(i -> i

    * 2) .take(2) .defaultIfEmpty(0) .subscribe(); Opérateur defaultIfEmpty
  43. @BriceDutheil @dwursteisen #DevoxxFR #Rx Question : Comment convertir une liste

    de mots en minuscules et filtrer pour garder uniquement les mots de plus de 3 lettres ?
  44. @BriceDutheil @dwursteisen #DevoxxFR #Rx Concept de Monade Une structure ayant

    - Un type paramétré (Observable<T>) - Une factory (Observable.just(...)) - Une méthode flatMap (obs.flatMap(...)) - Structure programmatique (Grossièrement)
  45. @BriceDutheil @dwursteisen #DevoxxFR #Rx Structure programmatique List<String> words = Stream.of("bonJour",

    "bonjour", "Hello") .map(str -> str.toLowerCase()) .distinct() .collect(Collectors.toList()); Factory
  46. @BriceDutheil @dwursteisen #DevoxxFR #Rx Structure programmatique List<String> words = Stream.of("bonJour",

    "bonjour", "Hello") .map(str -> str.toLowerCase()) .distinct() .collect(Collectors.toList()); Étape 1
  47. @BriceDutheil @dwursteisen #DevoxxFR #Rx Structure programmatique List<String> words = Stream.of("bonJour",

    "bonjour", "Hello") .map(str -> str.toLowerCase()) .distinct() .collect(Collectors.toList()); Étape 2
  48. @BriceDutheil @dwursteisen #DevoxxFR #Rx Structure programmatique List<String> words = Stream.of("bonJour",

    "bonjour", "Hello") .map(str -> str.toLowerCase()) .distinct() .collect(Collectors.toList()); Étape 3 (fonction terminale)
  49. @BriceDutheil @dwursteisen #DevoxxFR #Rx Évitez les effets de bord Structure

    programmatique une fonction est dite à effet de bord si elle modifie un état autre que sa valeur de retour
  50. @BriceDutheil @dwursteisen #DevoxxFR #Rx Évitez les effets de bord Structure

    programmatique une fonction est dite à effet de bord si elle modifie un état autre que sa valeur de retour (Comme avec les streams Java 8)
  51. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord List<Integer> result =

    new ArrayList<>(); Observable.just(1, 2, 3) .map(i -> result.add(i)) .subscribe();
  52. @BriceDutheil @dwursteisen #DevoxxFR #Rx List<Integer> result = new ArrayList<>(); Observable.just(1,

    2, 3) .map(i -> result.add(i)) .subscribe(); Effets de bord Effet de bord
  53. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord Observable.just(1, 2, 3)

    .map(i -> { LOGGER.info("value : " + i); return i; }) .subscribe();
  54. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .map(i -> {

    LOGGER.info("value : " + i); return i; }) .subscribe(); Effets de bord Effet de bord
  55. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord Observable.just(1, 2, 3)

    // explicite .doOnNext(i -> LOGGER.info("Valeur :" + i)) .subscribe();
  56. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord Observable.just(1, 2, 3)

    .doOnSubscribe(() -> LOGGER.info("Subscribe")) .doOnNext(i -> LOGGER.info("Valeur :" + i)) .doOnTerminate(() -> LOGGER.info("Terminate")) .subscribe();
  57. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord Observable.just(1, 2, 3)

    .doOnSubscribe(() -> LOGGER.info("Subscribe")) .doOnNext(i -> LOGGER.info("Valeur :" + i)) .doOnTerminate(() -> LOGGER.info("Terminate")) .subscribe();
  58. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord Observable.just(1, 2, 3)

    .doOnSubscribe(() -> LOGGER.info("Subscribe")) .doOnNext(i -> LOGGER.info("Valeur :" + i)) .doOnTerminate(() -> LOGGER.info("Terminate")) .subscribe();
  59. @BriceDutheil @dwursteisen #DevoxxFR #Rx Effets de bord Observable.just(1, 2, 3)

    .doOnSubscribe(() -> LOGGER.info("Subscribe")) .doOnNext(i -> LOGGER.info("Valeur :" + i)) .doOnTerminate(() -> LOGGER.info("Terminate")) .subscribe();
  60. @BriceDutheil @dwursteisen #DevoxxFR #Rx // attention ! état partagé !

    AtomicLong errors = new AtomicLong(0); Observable.just(1, 2, 3) .doOnError(ex -> errors.incrementAndGet()) .subscribe();
  61. @BriceDutheil @dwursteisen #DevoxxFR #Rx // attention ! état partagé !

    AtomicLong errors = new AtomicLong(0); Observable.just(1, 2, 3) .doOnError(ex -> errors.incrementAndGet()) .subscribe();
  62. @BriceDutheil @dwursteisen #DevoxxFR #Rx Comment créer des Observables Observable.just() Observable.fromCallable()

    Observable.defer() Observable.using() Observable.create() Simple d’utilisation
  63. @BriceDutheil @dwursteisen #DevoxxFR #Rx Comment créer des Observables Observable.just() Observable.fromCallable()

    Observable.defer() Observable.using() Observable.create() Complexe d’utilisation
  64. @BriceDutheil @dwursteisen #DevoxxFR #Rx Comment créer des Observables Observable.just() Observable.fromCallable()

    Observable.defer() Observable.using() Observable.create() Complexe d’utilisation Offre un maximum de contrôle
  65. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe();
  66. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe(); Création d’une ressource
  67. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe(); Émission d’un résultat
  68. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe(); Fermeture de la ressource
  69. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { subscriber.add(/* ... */);

    subscriber.setProducer(n -> { subscriber.onNext(/* ... */); // ... subscriber.onCompleted(); }); }).subscribe();
  70. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { subscriber.add(/* ... */);

    subscriber.setProducer(n -> { subscriber.onNext(/* ... */); // ... subscriber.onCompleted(); }); }).subscribe(); ? ? ?
  71. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { subscriber.add(/* ... */);

    subscriber.setProducer(n -> { subscriber.onNext(/* ... */); // ... subscriber.onCompleted(); }); }).subscribe(); Désabonnement Backpressure Contrat Rx
  72. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { for (int i

    = 0; i < 10; i++) { if (!subscriber.isUnsubscribed()) { subscriber.onNext(compute(i)); } } subscriber.onCompleted(); }).subscribe();
  73. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { for (int i

    = 0; i < 10; i++) { if (!subscriber.isUnsubscribed()) { subscriber.onNext(compute(i)); } } subscriber.onCompleted(); }).subscribe(); Le flux est-il encore actif ? Éviter de lancer ce calcul (coûteux)
  74. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { DB db =

    new DB(); subscriber.add(Subscriptions.create(() -> db.closeDb())); subscriber.onNext(db); subscriber.onCompleted(); }).subscribe();
  75. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { DB db =

    new DB(); subscriber.add(Subscriptions.create(() -> db.closeDb())); subscriber.onNext(db); subscriber.onCompleted(); }).subscribe(); Code de désabonnement
  76. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe(); Plus simple !
  77. @BriceDutheil @dwursteisen #DevoxxFR #Rx On a vu Tomcat et les

    servlets OS Tomcat / BIO client Requête # Acceptor Thread Thread war
  78. @BriceDutheil @dwursteisen #DevoxxFR #Rx Par défaut synchrone* method() { Observable.from(items)

    .map(this::transform) .flatMap(this::computeRelativeItems) .subscribe(response::resume, response::resume); } Travail sur le Thread courant
  79. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxJava permet un contrôle explicite de

    la concurrence avec observeOn() ou avec subscribeOn()
  80. @BriceDutheil @dwursteisen #DevoxxFR #Rx method() { Observable.from(items) .observeOn( ) .map(this::transform)

    .flatMap(this::computeRelativeItems) .subscribe(response::resume, response::resume); } Thread courant
  81. @BriceDutheil @dwursteisen #DevoxxFR #Rx method() { Observable.from(items) .observeOn( ) .map(this::transform)

    .flatMap(this::computeRelativeItems) .subscribe(response::resume, response::resume); } Thread courant
  82. @BriceDutheil @dwursteisen #DevoxxFR #Rx Par défaut synchrone* method() { Observable.from(items)

    .map(this::transform) .delay(10, SECONDS)) .subscribe(response::resume, response::resume); } Travail sur le Thread courant * Certains opérateurs fonctionnent par défaut sur un autre Scheduler
  83. @BriceDutheil @dwursteisen #DevoxxFR #Rx Par défaut synchrone* method() { Observable.from(items)

    .map(this::transform) .delay(10, SECONDS)) .subscribe(response::resume, response::resume); } Travail sur le Thread courant * Certains opérateurs fonctionnent par défaut sur un autre Scheduler Change le contexte
  84. @BriceDutheil @dwursteisen #DevoxxFR #Rx method() { Observable.from(items) .observeOn(Schedulers.computation()) .map(this::transform) .flatMap(this::computeRelativeItems)

    .subscribeOn( ) .subscribe(response::resume, response::resume); } Thread courant Travail planifié sur computation
  85. @BriceDutheil @dwursteisen #DevoxxFR #Rx method() { Observable.from(items) .observeOn(Schedulers.computation()) .map(this::transform) .flatMap(this::computeRelativeItems)

    .subscribeOn( ) .subscribe(response::resume, response::resume); } Thread courant Travail planifié sur computation
  86. @BriceDutheil @dwursteisen #DevoxxFR #Rx method() { Observable.from(items) .observeOn(Schedulers.computation()) .map(this::transform) .flatMap(this::computeRelativeItems)

    .subscribeOn(Schedulers.io()) .subscribe(response::resume, response::resume); } Thread courant Travail planifié sur computation Production des éléments sur io
  87. @BriceDutheil @dwursteisen #DevoxxFR #Rx Il ne peut y avoir qu’un

    seul subscribeOn() Premier arrivé, premier servi !
  88. @BriceDutheil @dwursteisen #DevoxxFR #Rx computation() .map(str -> …) Élément #

    .filter(…) Élément <> .merge(…) Élément ‘a’ .map(str -> …)
  89. @BriceDutheil @dwursteisen #DevoxxFR #Rx computation() Nombre de coeur CPU .map(str

    -> …) Élément # .filter(…) Élément <> .merge(…) Élément ‘a’ .map(str -> …)
  90. @BriceDutheil @dwursteisen #DevoxxFR #Rx computation() Nombre de coeur CPU .debounce(…)

    .filter(e -> ...) .merge(...) Élément # .filter(…) Élément <> .merge(…)
  91. @BriceDutheil @dwursteisen #DevoxxFR #Rx io() DB Network Internet Élément #

    .filter(…) Élément <> .merge(…) Élément ‘a’ .flatMap(…)
  92. @BriceDutheil @dwursteisen #DevoxxFR #Rx io() DB Network Internet Élément #

    .filter(…) Élément <> .merge(…) Élément ‘a’ .flatMap(…)
  93. @BriceDutheil @dwursteisen #DevoxxFR #Rx io() DB Network Internet Élément #

    .filter(…) Élément <> .merge(…) Élément ‘a’ .flatMap(…)
  94. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public Response wait(@PathParam("heavyness") int heavyness) throws InterruptedException { System.out.format("heavy work that will take %dms%n", heavyness); MILLISECONDS.sleep(heavyness); // hard and long work return Response.ok(format("Waited %dms", heavyness)).build(); } }
  95. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public Response wait(@PathParam("heavyness") int heavyness) throws InterruptedException { System.out.format("heavy work that will take %dms%n", heavyness); MILLISECONDS.sleep(heavyness); // hard and long work return Response.ok(format("Waited %dms", heavyness)).build(); } } Travail
  96. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { executor.submit(() -> { try { MILLISECONDS.sleep(heavyness); ar.resume(format("Waited %dms", heavyness)); } catch (InterruptedException e) { Thread.interrupted(); ar.resume(e); } }); } }
  97. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { executor.submit(() -> { try { MILLISECONDS.sleep(heavyness); ar.resume(format("Waited %dms", heavyness)); } catch (InterruptedException e) { Thread.interrupted(); ar.resume(e); } }); } } JAX-RS 2.0 (Servlet 3.0)
  98. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { executor.submit(() -> { try { MILLISECONDS.sleep(heavyness); ar.resume(format("Waited %dms", heavyness)); } catch (InterruptedException e) { Thread.interrupted(); ar.resume(e); } }); } } JAX-RS 2.0 (Servlet 3.0) Tache asynchrone
  99. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { executor.submit(() -> { try { MILLISECONDS.sleep(heavyness); ar.resume(format("Waited %dms", heavyness)); } catch (InterruptedException e) { Thread.interrupted(); ar.resume(e); } }); } } JAX-RS 2.0 (Servlet 3.0) thread pool Tache asynchrone
  100. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { executor.submit(() -> { try { MILLISECONDS.sleep(heavyness); ar.resume(format("Waited %dms", heavyness)); } catch (InterruptedException e) { Thread.interrupted(); ar.resume(e); } }); } } JAX-RS 2.0 (Servlet 3.0) thread pool Tache asynchrone response.resume() pour retourner la réponse HTTP
  101. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } }
  102. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } } Traitement asynchrone avec Rx
  103. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } } Prend les paramètres
  104. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } } Scheduler sur lequel faire le traitement ≈≈> executor
  105. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } } Sera exécuté sur un Computation thread Thread de la servlet
  106. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } } Travail
  107. @BriceDutheil @dwursteisen #DevoxxFR #Rx @Path("heavy-stuff") public static class WaitResource {

    @GET @Path("{heavyness}") public void wait(@Suspended AsyncResponse ar, @PathParam("heavyness") int heavyness) { response.setTimeout(50, MILLISECONDS); Observable.just(waitTime) .observeOn(Schedulers.computation()) .map(this::longProcess) .subscribe(result -> ar.resume(result), throwable -> ar.resume(throwable)); } private String longProcess(int heavyness) { ... } } onNext : résultat du traitement onError : termine la requête HTTP avec une erreur
  108. @BriceDutheil @dwursteisen #DevoxxFR #Rx Introduire un TimeoutHandler Status HTTP Headers

    Entité personnalisée Arrêt / interruption des traitements ...
  109. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxNetty.createHttpServer(port, (request, response) -> { if

    (request.getHttpMethod() == HttpMethod.POST) { return request.getContent() .map(TehServer::parseBody) .map(TehServer::longStuff) .flatMap(result -> { response.setStatus(HttpResponseStatus.OK); response.writeString(result); return response.close(); }).ignoreElements(); } return response.close(false); });
  110. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxNetty.createHttpServer(port, (request, response) -> { if

    (request.getHttpMethod() == HttpMethod.POST) { return request.getContent() .map(TehServer::parseBody) .map(TehServer::longStuff) .flatMap(result -> { response.setStatus(HttpResponseStatus.OK); response.writeString(result); return response.close(); }).ignoreElements(); } return response.close(false); }); Créer un serveur
  111. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxNetty.createHttpServer(port, (request, response) -> { if

    (request.getHttpMethod() == HttpMethod.POST) { return request.getContent() .map(TehServer::parseBody) .map(TehServer::longStuff) .flatMap(result -> { response.setStatus(HttpResponseStatus.OK); response.writeString(result); return response.close(); }).ignoreElements(); } return response.close(false); }); Handler pour gérer la requête
  112. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxNetty.createHttpServer(port, (request, response) -> { if

    (request.getHttpMethod() == HttpMethod.POST) { return request.getContent() .map(TehServer::parseBody) .map(TehServer::longStuff) .flatMap(result -> { response.setStatus(HttpResponseStatus.OK); response.writeString(result); return response.close(); }).ignoreElements(); } return response.close(false); }); Retourne un Observable
  113. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxNetty.createHttpServer(port, (request, response) -> { if

    (request.getHttpMethod() == HttpMethod.POST) { return request.getContent() .map(TehServer::parseBody) .map(TehServer::longStuff) .flatMap(result -> { response.setStatus(HttpResponseStatus.OK); response.writeString(result); return response.close(); }).ignoreElements(); } return response.close(false); }); Travail sur la request avec RxJava
  114. @BriceDutheil @dwursteisen #DevoxxFR #Rx … mais demande un effort appréciable

    sur la mise en oeuvre des aspects d’une API HTTP publique
  115. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .subscribe(response::resume); } Un Observable qui renvoie quelque chose sur AsyncResponse
  116. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { … }) .subscribe(response::resume); } Ici seront fait les appels
  117. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Response aResponse = wsA(lastNumber); … }) .subscribe(response::resume); } Appel au service A
  118. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<Response> aObs = Observable.fromCallable(() -> wsA(lastNumber)); … }) .subscribe(response::resume); } Appel observable au service A
  119. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsA(lastNumber)) .map(r -> r.readEntity(String.class)); … }) .subscribe(response::resume); } Transformation de la payload
  120. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsA(lastNumber)) .map(r -> r.readEntity(String.class)); Observable<String> bObs = Observable.fromCallable(() -> wsB(lastNumber)) .map(r -> r.readEntity(String.class)); … }) .subscribe(response::resume); } Appel observable au service B
  121. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsA(lastNumber)) .map(r -> r.readEntity(String.class)); Observable<String> bObs = Observable.fromCallable(() -> wsB(lastNumber)) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume); } Composition de A et de B
  122. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsA(lastNumber)) .map(r -> r.readEntity(String.class)); Observable<String> bObs = Observable.fromCallable(() -> wsB(lastNumber)) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, response::resume); } Ajout d’un handler onError
  123. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsA(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); Observable<String> bObs = Observable.fromCallable(() -> wsB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, response::resume); } Appels réseau exécutés sur des threads dédiés aux I/O
  124. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsA(lastNumber)) .subscribeOn(Schedulers.io()) .flatMap(r -> { if (r.getStatus() == 503) { return Observable.error( new ServiceUnavailable("service A")); } return just(r.readEntity(String.class)); }); Observable<String> bObs = Observable.fromCallable(() -> wsB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, response::resume); } Gestion du statut HTTP 503
  125. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsServiceA(lastNumber)) .subscribeOn(Schedulers.io()) .flatMap(r -> { if (r.getStatus() != 200) { return Observable.error( new ServiceUnavailable("service A")); } return Observable.just(r.readEntity(String.class)); }); Observable<String> bObs = Observable.fromCallable(() -> wsServiceB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(error); }); } AsyncReponse.resume(Throwable) ⇒ 500 Server Error
  126. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsServiceA(lastNumber)) .subscribeOn(Schedulers.io()) .flatMap(r -> { if (r.getStatus() != 200) { return Observable.error(new ServiceUnavailable("service A")); } return Observable.just(r.readEntity(String.class)); }); Observable<String> bObs = Observable.fromCallable(() -> wsServiceB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(Response.serverError() .entity(error.getMessage()) .build()); }); } Personnalisation de la payload d’erreur
  127. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsServiceA(lastNumber)) .subscribeOn(Schedulers.io()) .flatMap(r -> { if (r.getStatus() != 200) { return Observable.error(new ServiceUnavailable("service A")); } return Observable.just(r.readEntity(String.class)); }); Observable<String> bObs = Observable.fromCallable(() -> wsServiceB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(Response.status(((HttpServiceEx) error).httpStatus) .entity(error.getMessage()) .build()); }); } Propagation du statut HTTP
  128. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeInterestingValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsServiceA(lastNumber)) .subscribeOn(Schedulers.io()) .flatMap(r -> { if (r.getStatus() != 200) { return Observable.error(new ServiceUnavailable("service A")); } return Observable.just(r.readEntity(String.class)); }) .retry(2); Observable<String> bObs = Observable.fromCallable(() -> wsServiceB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(Response.status(((HttpServiceException) error).proposedHttpStatus) .entity(error.getMessage()) .build()); }); } Recommencera 2 fois
  129. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> makeValue()) .flatMap(lastNumber -> { Observable<String> aObs = Observable.fromCallable(() -> wsServiceA(lastNumber)) .subscribeOn(Schedulers.io()) .flatMap(r -> { if (r.getStatus() != 200) { return Observable.error(new ServiceUnavailable("service A")); } return Observable.just(r.readEntity(String.class)); }) .retry(2); Observable<String> bObs = Observable.fromCallable(() -> wsServiceB(lastNumber)) .subscribeOn(Schedulers.io()) .map(r -> r.readEntity(String.class)) .timeout(500, MILLISECONDS, error(new ServiceTimeout("service B"))); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(Response.status(((HttpServiceException) error).proposedHttpStatus) .entity(error.getMessage()) .build()); }); } Défini un délai acceptable
  130. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> new SecureRandom().nextInt(1000)) .flatMap(lastNumber -> { Observable<String> aObs = new ServiceACommand(lastNumber).observe(); Observable<String> bObs = new ServiceBCommand(lastNumber).observe(); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(Response.statusError() .entity(error.getMessage()) .build()); }); } Commandes Hystrix
  131. @BriceDutheil @dwursteisen #DevoxxFR #Rx @GET public void wait(@Suspended final AsyncResponse

    response) { Observable.fromCallable(() -> new SecureRandom().nextInt(1000)) .flatMap(lastNumber -> { Observable<String> aObs = new ServiceACommand(lastNumber).observe(); Observable<String> bObs = new ServiceBCommand(lastNumber).observe(); return Observable.zip(aObs, bObs, (a, b) -> format("=> Result : %s, %s", a, b); }) .subscribe(response::resume, (error) -> { response.resume(Response.serverError() .entity(error.getMessage()) .build()); }); } Erreur 500
  132. @BriceDutheil @dwursteisen #DevoxxFR #Rx class ServiceACommand extends HystrixObservableCommand<String> { private

    int number; protected ServiceACommand(int number) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("work group")) .andCommandKey(HystrixCommandKey.Factory.asKey("work")) .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(500))); this.number = number; } @Override protected Observable<String> construct() { return Observable.fromCallable(() -> wsServiceB(number)) .map(response -> response.readEntity(String.class)); } @Override protected Observable<String> resumeWithFallback() { return Observable.just("Service A too long"); } }
  133. @BriceDutheil @dwursteisen #DevoxxFR #Rx class ServiceACommand extends HystrixObservableCommand<String> { private

    int number; protected ServiceACommand(int number) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("work group")) .andCommandKey(HystrixCommandKey.Factory.asKey("work")) .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(500))); this.number = number; } @Override protected Observable<String> construct() { return Observable.fromCallable(() -> wsServiceB(number)) .map(response -> response.readEntity(String.class)); } @Override protected Observable<String> resumeWithFallback() { return Observable.just("Service A too long"); } } Commande Hystrix Type de retour
  134. @BriceDutheil @dwursteisen #DevoxxFR #Rx class ServiceACommand extends HystrixObservableCommand<String> { private

    int number; protected ServiceACommand(int number) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("work group")) .andCommandKey(HystrixCommandKey.Factory.asKey("work")) .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(500))); this.number = number; } @Override protected Observable<String> construct() { return Observable.fromCallable(() -> wsServiceB(number)) .map(response -> response.readEntity(String.class)); } @Override protected Observable<String> resumeWithFallback() { return Observable.just("Service A too long"); } } Commande Hystrix observable Construction
  135. @BriceDutheil @dwursteisen #DevoxxFR #Rx class ServiceACommand extends HystrixObservableCommand<String> { private

    int number; protected ServiceACommand(int number) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("work group")) .andCommandKey(HystrixCommandKey.Factory.asKey("work")) .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(500))); this.number = number; } @Override protected Observable<String> construct() { return Observable.fromCallable(() -> wsServiceB(number)) .map(response -> response.readEntity(String.class)); } @Override protected Observable<String> resumeWithFallback() { return Observable.just("Service A too long"); } } Appel observable
  136. @BriceDutheil @dwursteisen #DevoxxFR #Rx class ServiceACommand extends HystrixObservableCommand<String> { private

    int number; protected ServiceACommand(int number) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("work group")) .andCommandKey(HystrixCommandKey.Factory.asKey("work")) .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(500))); this.number = number; } @Override protected Observable<String> construct() { return Observable.fromCallable(() -> wsServiceB(number)) .map(response -> response.readEntity(String.class)); } @Override protected Observable<String> resumeWithFallback() { return Observable.just("Service A too long"); } } Plan de secours En cas d’erreur
  137. @BriceDutheil @dwursteisen #DevoxxFR #Rx class ServiceACommand extends HystrixObservableCommand<String> { private

    int number; protected ServiceACommand(int number) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("work group")) .andCommandKey(HystrixCommandKey.Factory.asKey("work")) .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(500))); this.number = number; } @Override protected Observable<String> construct() { return Observable.fromCallable(() -> wsServiceB(number)) .map(response -> response.readEntity(String.class)); } @Override protected Observable<String> resumeWithFallback() { return Observable.just("Service A too long"); } } Personnalisation des paramètres de la commande
  138. @BriceDutheil @dwursteisen #DevoxxFR #Rx <T> void executeCommand(HystrixObservableCommand<T> command) { command.observe()

    .subscribe( result -> Response.ok(result).build(), throwable -> { if (throwable instanceof HystrixRuntimeException) { if (((HystrixRuntimeException) throwable).getFailureType() == FailureType.SHORTCIRCUIT) { Response.status(Status.SERVICE_UNAVAILABLE) .header(HttpHeaders.RETRY_AFTER, command.getProperties() .circuitBreakerSleepWindowInMilliseconds() .get() / 1000); } } Response.serverError().build(); }); } Pour une commande Hystrix
  139. @BriceDutheil @dwursteisen #DevoxxFR #Rx <T> void executeCommand(HystrixObservableCommand<T> command) { command.observe()

    .subscribe( result -> Response.ok(result).build(), throwable -> { if (throwable instanceof HystrixRuntimeException) { if (((HystrixRuntimeException) throwable).getFailureType() == FailureType.SHORTCIRCUIT) { Response.status(Status.SERVICE_UNAVAILABLE) .header(HttpHeaders.RETRY_AFTER, command.getProperties() .circuitBreakerSleepWindowInMilliseconds() .get() / 1000); } } Response.serverError().build(); }); } Si le problème vient de la commande Hystrix
  140. @BriceDutheil @dwursteisen #DevoxxFR #Rx private <T> void executeCommand(HystrixObservableCommand<T> command) {

    command.observe() .subscribe( result -> Response.ok(result).build(), throwable -> { if (throwable instanceof HystrixRuntimeException) { if (((HystrixRuntimeException) throwable).getFailureType() == FailureType.SHORTCIRCUIT) { Response.status(Status.SERVICE_UNAVAILABLE) .header(HttpHeaders.RETRY_AFTER, command.getProperties() .circuitBreakerSleepWindowInMilliseconds() .get() / 1000); } } Response.serverError().build(); }); } Et que le circuit est ouvert
  141. @BriceDutheil @dwursteisen #DevoxxFR #Rx private <T> void executeCommand(HystrixObservableCommand<T> command) {

    command.observe() .subscribe( result -> Response.ok(result).build(), throwable -> { if (throwable instanceof HystrixRuntimeException) { if (((HystrixRuntimeException) throwable).getFailureType() == FailureType.SHORTCIRCUIT) { Response.status(Status.SERVICE_UNAVAILABLE) .header(HttpHeaders.RETRY_AFTER, command.getProperties() .circuitBreakerSleepWindowInMilliseconds() .get() / 1000); } } Response.serverError().build(); }); } Alors proposer au client un prochain créneau
  142. @BriceDutheil @dwursteisen #DevoxxFR #Rx private <T> void executeCommand(HystrixObservableCommand<T> command) {

    command.observe() .subscribe( result -> Response.ok(result).build(), throwable -> { if (throwable instanceof HystrixRuntimeException) { if (((HystrixRuntimeException) throwable).getFailureType() == FailureType.SHORTCIRCUIT) { Response.status(Status.SERVICE_UNAVAILABLE) .header(HttpHeaders.RETRY_AFTER, command.getProperties() .circuitBreakerSleepWindowInMilliseconds() .get() / 1000); } } Response.serverError().build(); }); }
  143. @BriceDutheil @dwursteisen #DevoxxFR #Rx Hystrix demande une certaine dose de

    code Il est possible d’utiliser des annotations avec hystrix-javanica
  144. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3, 4) .toList() .map(list

    -> { return list.stream() .filter(i -> i % 2 == 0) .collect(Collectors.toList()); }).subscribe();
  145. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3, 4) .toList() .map(list

    -> { return list.stream() .filter(i -> i % 2 == 0) .collect(Collectors.toList()); }).subscribe();
  146. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3, 4) .toList() .map(list

    -> { return list.stream() .filter(i -> i % 2 == 0) .collect(Collectors.toList()); }).subscribe();
  147. @BriceDutheil @dwursteisen #DevoxxFR #Rx List<Observable<?>> obs = new ArrayList<>(); obs.add(Observable.defer(()

    -> Observable.just(1, 2, 3))); obs.add(Observable.just(4, 5, 6)); obs.add(Observable.fromCallable(() -> 7)); Observable.merge(obs).subscribe();
  148. @BriceDutheil @dwursteisen #DevoxxFR #Rx List<Observable<?>> obs = new ArrayList<>(); obs.add(Observable.defer(()

    -> Observable.just(1, 2, 3))); obs.add(Observable.just(4, 5, 6)); obs.add(Observable.fromCallable(() -> 7)); Observable.merge(obs).subscribe();
  149. @BriceDutheil @dwursteisen #DevoxxFR #Rx List<Observable<?>> obs = new ArrayList<>(); obs.add(Observable.defer(()

    -> Observable.just(1, 2, 3))); obs.add(Observable.just(4, 5, 6)); obs.add(Observable.fromCallable(() -> 7)); Observable.merge(obs).subscribe();
  150. @BriceDutheil @dwursteisen #DevoxxFR #Rx List<Observable<?>> obs = new ArrayList<>(); obs.add(Observable.defer(()

    -> Observable.just(1, 2, 3))); obs.add(Observable.just(4, 5, 6)); obs.add(Observable.fromCallable(() -> 7)); Observable.merge(obs).subscribe();
  151. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<Feed> feeds = Observable.merge( services.stream() .map(service

    -> service.newRequest().user(user).all().observe()) .map(feedObservable -> feedObservable .doOnError(throwable -> log.error(throwable.getMessage(), throwable)) .onErrorResumeNext(Observable.empty())) .collect(Collectors.toList())); feeds.subscribe();
  152. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<Feed> feeds = Observable.merge( services.stream() .map(service

    -> service.newRequest().user(user).all().observe()) .map(feedObservable -> feedObservable .doOnError(throwable -> log.error(throwable.getMessage(), throwable)) .onErrorResumeNext(Observable.empty())) .collect(Collectors.toList())); feeds.subscribe(); RxJava Stream Java 8
  153. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxView.clickEvents(tapButton) .doOnNext(new Action1<ViewClickEvent>() { @Override public

    void call(ViewClickEvent onClickEvent) { LOGGER.debug("--------- GOT A TAP"); } }) .map(new Func1<ViewClickEvent, Integer>() { @Override public Integer call(ViewClickEvent onClickEvent) { return 1; } }).subscribe();
  154. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxView.clickEvents(tapButton) .doOnNext(new Action1<ViewClickEvent>() { @Override public

    void call(ViewClickEvent onClickEvent) { LOGGER.debug("--------- GOT A TAP"); } }) .map(new Func1<ViewClickEvent, Integer>() { @Override public Integer call(ViewClickEvent onClickEvent) { return 1; } }).subscribe(); Code technique
  155. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe();
  156. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe();
  157. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe();
  158. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe();
  159. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe();
  160. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe(); Trop complexe !
  161. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> {

    Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }).subscribe(); Extraction
  162. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> transform(i))

    .subscribe(); public Observable<Object> transform(int i) { Observable<String> complex = webService.complexify(i) .map(str -> str.toLowerCase()) .retry(4); Observable<String> human = webService.toHumanString(i) .filter(str -> str.length() < 3) .take(1) .onErrorResumeNext(Observable.empty()) .defaultIfEmpty("Oups !"); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr); }
  163. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .flatMap(i -> transform(i))

    .subscribe(); public Observable<Object> transform(int i) { Observable<String> complex = fetchComplexify(i); Observable<String> human = fetchHuman(i); return Observable.zip(complex, human, (c, h) -> c + h) .flatMap(str -> webService.frequencyOf(str)) .flatMapIterable(itr -> itr) }
  164. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ne quittez pas la monade !

    Observable.just(1, 2, 3) .subscribe(arg -> { webservice.call(arg).subscribe(System.out::println); });
  165. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.just(1, 2, 3) .subscribe(arg -> {

    webservice.call(arg).subscribe(System.out::println); }); Ne quittez pas la monade ! Observable imbriqué
  166. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ne quittez pas la monade !

    Observable.just(1, 2, 3) .flatMap(arg -> webservice.call(arg)) .subscribe(System.out::println);
  167. @BriceDutheil @dwursteisen #DevoxxFR #Rx Ne quittez pas la monade !

    Observable.just(1, 2, 3) .flatMap(arg -> webservice.call(arg)) .subscribe(System.out::println);
  168. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(tps -> Observable.just(1, 2,

    3)) .subscribe(System.out::println, Throwable::printStackTrace, () -> System.out.println("Completed !"));
  169. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(tps -> Observable.just(1, 2,

    3)) .subscribe(System.out::println, Throwable::printStackTrace, () -> System.out.println("Completed !"));
  170. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(tps -> Observable.just(1, 2,

    3)) .subscribe(System.out::println, Throwable::printStackTrace, () -> System.out.println("Completed !"));
  171. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(tps -> Observable.just(1, 2,

    3)) .subscribe(System.out::println, Throwable::printStackTrace, () -> System.out.println("Completed !"));
  172. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(tps -> Observable.just(1, 2,

    3)) .subscribe(System.out::println, Throwable::printStackTrace, () -> System.out.println("Completed !"));
  173. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(tps -> Observable.just(1, 2,

    3)) .subscribe(System.out::println, Throwable::printStackTrace, () -> System.out.println("Completed !")); Cela n’arrivera jam ais !
  174. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { Timer timer =

    new Timer(); timer.onTick(() -> subscriber.onNext("tick")); }).subscribe();
  175. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { Timer timer =

    new Timer(); timer.onTick(() -> subscriber.onNext("tick")); }).subscribe();
  176. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { Timer timer =

    new Timer(); timer.onTick(() -> subscriber.onNext("tick")); }).subscribe();
  177. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { Timer timer =

    new Timer(); timer.onTick(() -> subscriber.onNext("tick")); }).subscribe(); Cold
  178. @BriceDutheil @dwursteisen #DevoxxFR #Rx Timer timer = new Timer(); Observable.create(subscriber

    -> { timer.onTick(() -> subscriber.onNext("tick")); }).subscribe();
  179. @BriceDutheil @dwursteisen #DevoxxFR #Rx Timer timer = new Timer(); Observable.create(subscriber

    -> { timer.onTick(() -> subscriber.onNext("tick")); }).subscribe();
  180. @BriceDutheil @dwursteisen #DevoxxFR #Rx Timer timer = new Timer(); Observable.create(subscriber

    -> { timer.onTick(() -> subscriber.onNext("tick")); }).subscribe();
  181. @BriceDutheil @dwursteisen #DevoxxFR #Rx Timer timer = new Timer(); Observable.create(subscriber

    -> { timer.onTick(() -> subscriber.onNext("tick")); }).subscribe(); Hot
  182. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<String> response = ws.call(); response .map(str

    -> str.toLowerCase()) .subscribe(System.out::println); response .subscribe(System.out::println);
  183. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<String> response = ws.call(); response .map(str

    -> str.toLowerCase()) .subscribe(System.out::println); response .subscribe(System.out::println);
  184. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<String> response = ws.call(); response .map(str

    -> str.toLowerCase()) .subscribe(System.out::println); response .subscribe(System.out::println);
  185. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<String> response = ws.call(); response .map(str

    -> str.toLowerCase()) .subscribe(System.out::println); response .subscribe(System.out::println); Cold
  186. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<String> response = ws.call(); response .map(str

    -> str.toLowerCase()) .subscribe(System.out::println); response .subscribe(System.out::println); Cold Création d’un appel réseau
  187. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable<String> response = ws.call(); response .map(str

    -> str.toLowerCase()) .subscribe(System.out::println); response .subscribe(System.out::println); Cold Création d’un second appel réseau Création d’un appel réseau
  188. @BriceDutheil @dwursteisen #DevoxxFR #Rx WARN AAA [2014-11-10 02:57:24,566] first_observable: 9

    MDC.put("ID", "AAA"); LOG.warning("first_observable: {}", l);
  189. @BriceDutheil @dwursteisen #DevoxxFR #Rx WARN AAA [2014-11-10 02:57:24,566] first_observable: 9

    MDC.put("ID", "AAA"); LOG.warning("first_observable: {}", l); Utilisation de ThreadLocal
  190. @BriceDutheil @dwursteisen #DevoxxFR #Rx WARN AAA [2014-11-10 02:57:24,566] first_observable: 9

    WARN BBB [2014-11-10 02:57:24,576] second_observable: 1 WARN AAA [2014-11-10 02:57:25,566] first_observable: 10 WARN CCC [2014-11-10 02:57:25,576] third_observable: 2 Contexte synchrone
  191. @BriceDutheil @dwursteisen #DevoxxFR #Rx WARN AAA [2014-11-10 02:57:24,566] first_observable: 9

    WARN BBB [2014-11-10 02:57:24,576] second_observable: 1 WARN AAA [2014-11-10 02:57:25,566] first_observable: 10 WARN CCC [2014-11-10 02:57:25,576] third_observable: 2 Contexte synchrone
  192. @BriceDutheil @dwursteisen #DevoxxFR #Rx WARN AAA [2014-11-10 02:57:24,566] first_observable: 9

    WARN BBB [2014-11-10 02:57:24,576] third_observable: 1 WARN AAA [2014-11-10 02:57:25,566] second_observable: 10 WARN BBB [2014-11-10 02:57:25,576] third_observable: 2 Contexte asynchrone
  193. @BriceDutheil @dwursteisen #DevoxxFR #Rx WARN AAA [2014-11-10 02:57:24,566] first_observable: 9

    WARN BBB [2014-11-10 02:57:24,576] third_observable: 1 WARN AAA [2014-11-10 02:57:25,566] second_observable: 10 WARN BBB [2014-11-10 02:57:25,576] third_observable: 2 Contexte asynchrone Interleaving
  194. @BriceDutheil @dwursteisen #DevoxxFR #Rx Il est nécessaire de remettre en

    place le contexte lors d’un changement de contexte.
  195. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxJavaPlugins.getInstance().registerSchedulersHook(new RxJavaSchedulersHook() { @Override public Action0

    onSchedule(Action0 action) { return super.onSchedule(() -> { MDC.setContextMap(mdcContextMap); try { action.call(); } finally { MDC.clear(); } }); } }); http://blog.mabn.pl/2014/11/rxjava-logback-and-mdc-threadlocal.html (version courte)
  196. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxJavaPlugins.getInstance().registerSchedulersHook(new RxJavaSchedulersHook() { @Override public Action0

    onSchedule(Action0 action) { return super.onSchedule(() -> { MDC.setContextMap(mdcContextMap); try { action.call(); } finally { MDC.clear(); } }); } }); http://blog.mabn.pl/2014/11/rxjava-logback-and-mdc-threadlocal.html (version courte) À chaque planification de tâche
  197. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxJavaPlugins.getInstance().registerSchedulersHook(new RxJavaSchedulersHook() { @Override public Action0

    onSchedule(Action0 action) { return super.onSchedule(() -> { MDC.setContextMap(mdcContextMap); try { action.call(); } finally { MDC.clear(); } }); } }); http://blog.mabn.pl/2014/11/rxjava-logback-and-mdc-threadlocal.html (version courte) À chaque planification de tâche On planifie une autre tâche
  198. @BriceDutheil @dwursteisen #DevoxxFR #Rx RxJavaPlugins.getInstance().registerSchedulersHook(new RxJavaSchedulersHook() { @Override public Action0

    onSchedule(Action0 action) { return super.onSchedule(() -> { MDC.setContextMap(mdcContextMap); try { action.call(); } finally { MDC.clear(); } }); } }); http://blog.mabn.pl/2014/11/rxjava-logback-and-mdc-threadlocal.html (version courte) À chaque planification de tâche On planifie une autre tâche Qui remet le contexte et qui exécute notre tâche
  199. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(i -> { try

    { RandomException.mayThrowException(); return Observable.empty(); } catch (IOException e) { return Observable.error(e); } }).subscribe(System.out::println, e -> { /* ... */ });
  200. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(i -> { try

    { RandomException.mayThrowException(); return Observable.empty(); } catch (IOException e) { return Observable.error(e); } }).subscribe(System.out::println, e -> { /* ... */ }); Quelle valeur a généré l’exception ?
  201. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(i -> { try

    { RandomException.mayThrowException(); return Observable.empty(); } catch (IOException e) { return Observable.error(OnErrorThrowable.addValueAsLastCause(e, i)); } }).subscribe(System.out::println, e -> { /* ... */ }); Associe l’exception avec une valeur
  202. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.interval(1, TimeUnit.SECONDS) .flatMap(i -> { try

    { RandomException.mayThrowException(); return Observable.empty(); } catch (IOException e) { return Observable.error(OnErrorThrowable.addValueAsLastCause(e, i)); } }).subscribe(System.out::println, e -> { Object value = OnErrorThrowable.from(e).getValue(); }); Récupère la valeur depuis l’exception
  203. @BriceDutheil @dwursteisen #DevoxxFR #Rx public interface WebService { Observable<String> call();

    } Va émettre une valeur ? Va émettre plusieurs valeurs ? Va émettre aucune valeur ?
  204. @BriceDutheil @dwursteisen #DevoxxFR #Rx public interface WebService { Observable<String> list();

    Single<String> get(); Completable perform(); } Va émettre un ensemble de valeurs va émettre une unique valeur ne va pas émettre de valeur
  205. @BriceDutheil @dwursteisen #DevoxxFR #Rx Subscription flux1 = Observable.interval(1, TimeUnit.SECONDS) .subscribe(System.out::println);

    Subscription flux2 = Observable.interval(100, TimeUnit.MILLISECONDS) .subscribe(System.err::println);
  206. @BriceDutheil @dwursteisen #DevoxxFR #Rx Subscription flux1 = Observable.interval(1, TimeUnit.SECONDS) .subscribe(System.out::println);

    Subscription flux2 = Observable.interval(100, TimeUnit.MILLISECONDS) .subscribe(System.err::println); flux1.unsubscribe(); flux2.unsubscribe();
  207. @BriceDutheil @dwursteisen #DevoxxFR #Rx Subscription flux1 = Observable.interval(1, TimeUnit.SECONDS) .subscribe(System.out::println);

    Subscription flux2 = Observable.interval(100, TimeUnit.MILLISECONDS) .subscribe(System.err::println); CompositeSubscription subscription = new CompositeSubscription(); subscription.add(flux1); subscription.add(flux2); Compose les abonnements
  208. @BriceDutheil @dwursteisen #DevoxxFR #Rx Subscription flux1 = Observable.interval(1, TimeUnit.SECONDS) .subscribe(System.out::println);

    Subscription flux2 = Observable.interval(100, TimeUnit.MILLISECONDS) .subscribe(System.err::println); CompositeSubscription subscription = new CompositeSubscription(); subscription.add(flux1); subscription.add(flux2); subscription.unsubscribe(); Désabonne les flux
  209. @BriceDutheil @dwursteisen #DevoxxFR #Rx private CompositeSubscription subscription; @Override public void

    onResume() { Subscription interval = Observable.interval(1, TimeUnit.SECONDS) .subscribe(/* ... */); Subscription click = clicks() .subscribe(/*...*/); subscription = new CompositeSubscription(); subscription.add(interval); subscription.add(click); } @Override public void onPause() { subscription.unsubscribe(); } Désabonne les flux
  210. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.create(subscriber -> { subscriber.add(new Subscription() {

    @Override public void unsubscribe() { // fermeture du flux } }); }); Fermeture des ressources, etc
  211. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe();
  212. @BriceDutheil @dwursteisen #DevoxxFR #Rx Observable.using( () -> new DB(), db

    -> Observable.just(db.doQuery()), db -> db.closeDb()) .subscribe(); Fermeture des ressources, etc
  213. @BriceDutheil @dwursteisen #DevoxxFR #Rx Les autres talks sur les Reactive

    Extensions - A lite Rx API for the JVM Hands On Labs - Mercredi 9h30 - Baptême du feu avec Rx Hands On Labs - Vendredi 11h00 Alexandre Victoor Florent Le Gall Sebastien Deleuze Stephane Maldini