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

Get Reactive With Project Reactor and Spring 5

Get Reactive With Project Reactor and Spring 5

Spring – the tool that is over 13 years pleases our hearts and eyes of customers. Despite this long history of development the project continues to follow the latest trends and provides to developers the awesome set of add-ons that simplify the daily life.

During this tech-talk, I would like to discuss delicacies which were cooked by developers from Pivotal and how Reactivity reflected on the 5th iteration of the framework.

Oleh Dokuka

May 26, 2017
Tweet

More Decks by Oleh Dokuka

Other Decks in Programming

Transcript

  1. • JSE at Levi9 • 6+ dev experience • Spring

    fan • Reactivity preacher /oleh.dokuka /OlegDokuka /OlehDokuka WHO AM I?
  2. Agenda 1. Why Reactive? 2. Reactivity Generations 3. Reactive Streams

    4. Project Reactor 5. Reactivity in Action (Project Reactor + Spring 5)
  3. Publisher Subscriber public interface Subscriber<T> { public void onSubscribe(Subscription s);

    public void onNext(T t); public void onError(Throwable t); public void onComplete(); }
  4. by

  5. @RestController @RequestMapping("/api/v1/statistics") public class StatisticResource { private final StatisticService statisticService;

    @Autowired public StatisticResource( StatisticService statisticService) { this.statisticService = statisticService; } @GetMapping("/users") public List<UsersStatisticVM> getUsersStatistic() { return statisticService.getUsersStatistic(); } }
  6. @RestController @RequestMapping("/api/v1/statistics") public class StatisticResource { private final StatisticService statisticService;

    @Autowired public StatisticResource( StatisticService statisticService) { this.statisticService = statisticService; } @GetMapping("/users") public List<UsersStatisticVM> getUsersStatistic() { return statisticService.getUsersStatistic(); } }
  7. @RestController @RequestMapping("/api/v1/statistics") public class StatisticResource { private final StatisticService statisticService;

    @Autowired public StatisticResource( StatisticService statisticService) { this.statisticService = statisticService; } @GetMapping("/users") public Flux<UsersStatisticVM> getUsersStatistic() { return Flux.fromIterable(statisticService .getUsersStatistic()); }
  8. @RestController @RequestMapping("/api/v1/statistics") public class StatisticResource { private final StatisticService statisticService;

    @Autowired public StatisticResource( StatisticService statisticService) { this.statisticService = statisticService; } @GetMapping("/users") public Flux<UsersStatisticVM> getUsersStatistic() { return Flux.fromIterable(statisticService .getUsersStatistic()); }
  9. interface ReactiveCrudRepository<T, ID> extends Repository<T, ID> <S extends T> Mono<S>

    save(S entity); <S extends T> Flux<S> saveAll(Iterable<S> entities); <S extends T> Flux<S> saveAll(Publisher<S> entityStream); Mono<T> findById(ID id); Mono<T> findById(Mono<ID> id); Mono<Boolean> existsById(ID id); Mono<Boolean> existsById(Mono<ID> id); Flux<T> findAll(); Flux<T> findAllById(Iterable<ID> ids); Flux<T> findAllById(Publisher<ID> idStream); Mono<Long> count(); Mono<Void> deleteById(ID id); Mono<Void> delete(T entity); Mono<Void> deleteAll(Iterable<? extends T> entities); Mono<Void> deleteAll(Publisher<? extends T> entityStream); Mono<Void> deleteAll(); }
  10. interface ReactiveCrudRepository<T, ID> extends Repository<T, ID> <S extends T> Mono<S>

    save(S entity); <S extends T> Flux<S> saveAll(Iterable<S> entities); <S extends T> Flux<S> saveAll(Publisher<S> entityStream); Mono<T> findById(ID id); Mono<T> findById(Mono<ID> id); Mono<Boolean> existsById(ID id); Mono<Boolean> existsById(Mono<ID> id); Flux<T> findAll(); Flux<T> findAllById(Iterable<ID> ids); Flux<T> findAllById(Publisher<ID> idStream); Mono<Long> count(); Mono<Void> deleteById(ID id); Mono<Void> delete(T entity); Mono<Void> deleteAll(Iterable<? extends T> entities); Mono<Void> deleteAll(Publisher<? extends T> entityStream); Mono<Void> deleteAll(); }
  11. .publishOn() public static <S, V> Mono<V> mono( Mono<S> input, Function<...>

    transformer ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); }
  12. .publishOn() public static <S, V> Mono<V> mono( Mono<S> input, Function<...>

    transformer ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); }
  13. .publishOn() public static <S, V> Mono<V> mono( Mono<S> input, Function<...>

    transformer ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); }
  14. .publishOn() public static <S, V> Mono<V> mono( Mono<S> input, Function<...>

    transformer ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); }
  15. .publishOn() public static <S, V> Mono<V> mono( Mono<S> input, Function<...>

    transformer ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); }
  16. .publishOn() public static <S, V> Mono<V> mono( Mono<S> input, Function<...>

    transformer ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); }
  17. public static <S, V> Mono<V> mono( Mono<S> input, Function<...> transformer

    ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); } .publishOn()
  18. public static <S, V> Mono<V> mono( Mono<S> input, Function<...> transformer

    ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); } .subscribeOn() MonoProcessor
  19. public static <S, V> Mono<V> mono( Mono<S> input, Function<...> transformer

    ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); } .subscribeOn() MonoProcessor
  20. public static <S, V> Mono<V> mono( Mono<S> input, Function<...> transformer

    ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); } MonoProcessor .subscribeOn() …subscribe();
  21. public static <S, V> Mono<V> mono( Mono<S> input, Function<...> transformer

    ) { MonoProcessor<V> monoProcessor = MonoProcessor.create(); input .publishOn(Schedulers.elastic()) .transform(transformer) .subscribe(monoProcessor); return monoProcessor .subscribeOn(Schedulers.elastic()); } …subscribe(); MonoProcessor .subscribeOn() …subscribe();
  22. public interface WebClient { UriSpec<RequestHeadersSpec<?>> get(); //omitted... static WebClient create()

    { return new DefaultWebClientBuilder().build(); } interface RequestHeadersSpec<S extends RequestHeadersSpec<S>> { //omitted... Mono<ClientResponse> exchange(); //omitted... } interface ResponseSpec { <T> Mono<T> bodyToMono(Class<T> bodyType); <T> Flux<T> bodyToFlux(Class<T> elementType); <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType); <T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> element } }
  23. public interface WebClient { UriSpec<RequestHeadersSpec<?>> get(); //omitted... static WebClient create()

    { return new DefaultWebClientBuilder().build(); } interface RequestHeadersSpec<S extends RequestHeadersSpec<S>> { //omitted... Mono<ClientResponse> exchange(); //omitted... } interface ResponseSpec { <T> Mono<T> bodyToMono(Class<T> bodyType); <T> Flux<T> bodyToFlux(Class<T> elementType); <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType); <T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> element } }
  24. @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource = Flux.from(

    gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); }
  25. @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource = Flux.from(

    gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); } .onBP()
  26. @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource = Flux.from(

    gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); } .onBP()
  27. .onBP() @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource =

    Flux.from( gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); }
  28. .onBP() @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource =

    Flux.from( gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); }
  29. .onBP() @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource =

    Flux.from( gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); }
  30. .onBP() @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource =

    Flux.from( gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); } …subscribe();
  31. .onBP() @Autowired public GitterService( GitterClient gitterClient ) { gitterMessageSource =

    Flux.from( gitterClient .getMessages(null) ) .onBackpressureBuffer() .publish(1) .autoConnect(0); } …subscribe(); …subscribe();
  32. @Override public Flux<MessageVM> latest() { return chatClient.stream() .transform(MessageMapper ::toViewModelUnits );

    } @Override public Flux<MessageVM> latest() { return chatClient.stream() .transform(MessageMapper ::toViewModelUnits ); } @Override public Flux<MessageVM> latest() { return chatClient.stream() .transform(MessageMapper ::toViewModelUnits ); } .transform()
  33. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); }
  34. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform()
  35. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform()
  36. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } Channel .materialize()
  37. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform()
  38. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform()
  39. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform()
  40. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform()
  41. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .materialize() .transform()
  42. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() .materialize()
  43. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } Channel .materialize()
  44. Channel @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker

    ) { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .materialize()
  45. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker )

    { this.chatClient = chatClient; Flux<Message> saved = messageRepository .saveAll(chatClient.stream() .transform(MessageMapper::toDomainUnits) ); messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .materialize() Channel
  46. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  47. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  48. .retry() Channel this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof

    USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  49. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  50. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  51. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  52. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  53. .retry() Channel this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof

    USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  54. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  55. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  56. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel
  57. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() Channel .dematerialize()
  58. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); Channel .retry() .dematerialize()
  59. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) .mergeWith .defer(statistic)
  60. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) .mergeWith .defer(statistic) …subscribe();
  61. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) .mergeWith .defer(statistic) …subscribe();
  62. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) …subscribe(); this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .mergeWith .defer(statistic)
  63. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) …subscribe();
  64. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) …subscribe();
  65. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .flatMap() .cache(1) …subscribe();
  66. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); …subscribe(); .flatMap() .cache(1)
  67. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); Channel .retry() .dematerialize()
  68. this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof USE) .dematerialize()

    .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1); .retry() .dematerialize() .flatMap()
  69. .retry() .dematerialize() .flatMap() this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err

    instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  70. .dematerialize() .flatMap() .cache(1) this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err

    instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  71. .dematerialize() .flatMap() .cache(1) this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err

    instanceof USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  72. .flatMap() .cache(1) this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof

    USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  73. .flatMap() .cache(1) this.statisticPublisher = messageBroker .channel("statisticChanged") .retry(err -> err instanceof

    USE) .dematerialize() .flatMap(s -> doGetUserStatistic()) .mergeWith(Mono .defer(this::doGetUserStatistic) ) .cache(1);
  74. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  75. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  76. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  77. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  78. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  79. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  80. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .defaultIfEmpty(…) .zip(…) .fromDirect(…)
  81. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .defaultIfEmpty(…) .zip(…) .fromDirect(…)
  82. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null
  83. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null
  84. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null
  85. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null null
  86. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null
  87. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null
  88. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…) null
  89. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .map(…) .defaultIfEmpty(…) .zip(…)
  90. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .defaultIfEmpty(…) .zip(…) .fromDirect(…)
  91. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .defaultIfEmpty(…) .zip(…) .fromDirect(…)
  92. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM> topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .defaultIfEmpty(…) .zip(…) .fromDirect(…)
  93. private Mono<UsersStatisticVM> doGetUserStatistic Flux<UserVM> topActiveUser = userRepository .findAllOrderedByActivityDesc(PageRequest.o .map(UserMapper::toViewModelUnits); Flux<UserVM>

    topMentionedUser = userRepository .findAllOrderedByMentionDesc(PageRequest.of .map(UserMapper::toViewModelUnits); return Mono.fromDirect( Flux.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER) UsersStatisticVM::new ) ); } .defaultIfEmpty(…) .zip(…) .fromDirect(…)
  94. public interface WebSocketSession { String getId(); HandshakeInfo getHandshakeInfo(); DataBufferFactory bufferFactory();

    Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); default Mono<Void> close() { return close(CloseStatus.NORMAL); } Mono<Void> close(CloseStatus status); WebSocketMessage textMessage(String payload); WebSocketMessage binaryMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); WebSocketMessage pingMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); WebSocketMessage pongMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); }
  95. public interface WebSocketSession { String getId(); HandshakeInfo getHandshakeInfo(); DataBufferFactory bufferFactory();

    Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); default Mono<Void> close() { return close(CloseStatus.NORMAL); } Mono<Void> close(CloseStatus status); WebSocketMessage textMessage(String payload); WebSocketMessage binaryMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); WebSocketMessage pingMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); WebSocketMessage pongMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); }
  96. public interface WebSocketSession { String getId(); HandshakeInfo getHandshakeInfo(); DataBufferFactory bufferFactory();

    Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); default Mono<Void> close() { return close(CloseStatus.NORMAL); } Mono<Void> close(CloseStatus status); WebSocketMessage textMessage(String payload); WebSocketMessage binaryMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); WebSocketMessage pingMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); WebSocketMessage pongMessage( Function<DataBufferFactory, DataBuffer> payloadFactory); }
  97. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge(

    messageService.latest(), statisticService .usersStatisticStream() ); } network .merge()
  98. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge(

    messageService.latest(), statisticService .usersStatisticStream() ); } network .merge()
  99. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge(

    messageService.latest(), statisticService .usersStatisticStream() ); } network .merge()
  100. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge(

    messageService.latest(), statisticService .usersStatisticStream() ); } network .merge()
  101. .merge() @Override public Publisher<?> handle( Publisher<Object> in ) { return

    Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network
  102. network .merge() @Override public Publisher<?> handle( Publisher<Object> in ) {

    return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); }
  103. network .merge() @Override public Publisher<?> handle( Publisher<Object> in ) {

    return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); }
  104. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge(

    messageService.latest(), statisticService .usersStatisticStream() ); } network .merge()
  105. Q&A

  106. Resources 1. Why Reactive? [http://www.oreilly.com/programming/ free/why-reactive.csp] 2. Operator Fusion Part

    1 [http://akarnokd.blogspot.com/ 2016/03/operator-fusion-part-1.html] 3. Reactive Streams Specification [http://www.reactive- streams.org/] 4. Learn Project Reactor [http://projectreactor.io/learn]