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

Get ready for java.util.concurrent.Flow!

Get ready for java.util.concurrent.Flow!

Slides from the presentation regarding Reactive Streams implementations and Reactive Streams standard introduced in Java 9. This presentation was shown during JDD Conference on the 3rd of October 2017. I've also presented this talk during JUG meetup in Kraków on 4th of December 2017, infoMEET conference in Wrocław on 16th of December 2017 and Hackin' Gliwice mini-conference in Gliwice on 15th of March 2018.

Piotr Wittchen

October 03, 2017
Tweet

More Decks by Piotr Wittchen

Other Decks in Programming

Transcript

  1. List<Integer> list = Arrays.asList(1, 2, 3, 4, 5); for (Integer

    n : list) { if (n % 2 == 0) { operate(n); } }
  2. List<Integer> list = Arrays.asList(1, 2, 3, 4, 5); for (Integer

    n : list) { if (n % 2 == 0) { operate(n); } }
  3. List<Integer> list = Arrays.asList( , , , , ); for

    (Integer n : list) { if (n % 2 == 0) { operate(n); } } 1 2 4 3 5
  4. 1 2 4 3 5 Flowable.fromArray( , , , ,

    ) .filter(n -> n % 2 == 0) .subscribe(n -> operate(n));
  5. Flowable.fromArray(1, 2, 3, 4, 5) .filter(n -> n % 2

    == 0) .subscribe(n -> operate(n));
  6. Observable.range(1, 10) .map(i -> i * 100) .doOnNext(i -> printNumberWithThreadInfo("emitting",

    i)) .map(i -> i * 10) .subscribeOn(Schedulers.computation()) .observeOn(Schedulers.newThread()) .subscribe(integer -> printNumberWithThreadInfo("received", integer));
  7. Observable.range(1, 10) .map(i -> i * 100) .doOnNext(i -> printNumberWithThreadInfo("emitting",

    i)) .map(i -> i * 10) .subscribeOn(Schedulers.computation()) .observeOn(Schedulers.newThread()) .subscribe(integer -> printNumberWithThreadInfo("received", integer)); observeOn(...) observeOn(...) subscribeOn(...)
  8. List<String> list = new ArrayList<>(); new Thread(() -> ((Callback) ()

    -> { list.add("one"); ((Callback) () -> { list.add("two"); ((Callback) () -> { list.add("three"); System.out.println(list.stream().collect(Collectors.joining(" "))); }).execute(); }).execute(); }).execute()).start(); • • • •
  9. Flowable .fromCallable(() -> "one") .flatMap(name -> Flowable.fromCallable(() -> name.concat(" two")))

    .flatMap(name -> Flowable.fromCallable(() -> name.concat(" three"))) .subscribe(System.out::println);
  10. flowable.subscribe(new Subscriber<Object>() { @Override public void onSubscribe(Subscription s) { }

    @Override public void onNext(Object o) { } @Override public void onError(Throwable throwable) { System.out.println(throwable.getMessage()); } @Override public void onComplete() { } }); Subscriber
  11. Map<Integer, String> words = new HashMap<Integer, String>(); Flowable .fromArray(1, 2,

    3, 4, 5) .flatMap(i -> Flowable.just(i * 2)) .map(words::get) .throttleWithTimeout(1, TimeUnit.SECONDS) .lastElement() .subscribe(System.out::println);
  12. List<Integer> list = new ArrayList<>(); ActorSystem actorSystem = ActorSystem.create(); ActorMaterializer

    mat = ActorMaterializer.create(actorSystem); Source<Integer, NotUsed> source = Source.range(1, 5); source.runForeach(list::add, mat);
  13. Flowable<Integer> flowable = Flowable.just(1, 2, 3, 4); Flux<Integer> flux =

    Flux.just(5, 6, 7, 8); flowable .mergeWith(flux) .sorted() .subscribe(System.out::println); io.reactivex.Flowable reactor.core.publisher.Flux org.reactivestreams.Publisher java.util.concurrent.Flow.Publisher
  14. Flow.Publisher<Integer> publisher = subscriber -> { … // to be

    implemented }; publisher.subscribe(new Flow.Subscriber<>() { @Override public void onSubscribe(Flow.Subscription subscription) { ... } @Override public void onNext(Integer item) { ... } @Override public void onError(Throwable throwable) { ... } @Override public void onComplete() { ... } }); Publisher Subscriber
  15. Just for a presentation demo it’s better to use production

    ready libraries has to pass TCK (Technology Compatibility Kit) tests to keep RS ecosystem healthy!
  16. • part of Java 9 language • interfaces for reactive

    applications without implementations • external production ready implementations RxJava Reactor Akka Stream • have to be compatible with TCK • data transformation • callback hell • asynchronous operations • efficient consumption of the hardware resources • error handling API • can be used anywhere • clean maintainable concise modern