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

How to build a crypto trading platform with Spr...

How to build a crypto trading platform with Spring 5 and Reactor 3

In the talk, we will look at Reactive approaches with Spring 5 and Reactor 3 and see how to build a Reactive System using Spring Reactive Stack in details. In turn, we will discuss the common business needs, where these techniques fit well and how they may help you in solving complex problems in the most elegant and efficient way.

During the talk, we are going to build a Reactive Crypto-Trading Platform.

The central point which will be covered during the talk:

* Analyze of how to build a simple WebSocket API for data transfer;
* Analyze of existing Crypto-Platforms and the ways of integration with them using Reactor 3 and Spring WebFlux;
* Implementation of Trading and a Customer’s Wallet features;
* Security in Reactive solution with Reactive Spring Security;

What will be skipped:

* Blockchain mechanism;
* Purchasing/Sales mechanism of crypto-currencies in detail;
* Kotlin, obviously :);

Throughout all the talk, we will try to understand whether Reactor 3 and New Reactive Spring 5 helps us or not in solving of common business needs. Try to find out where Reactor 3 shines brightly, what works in WebFlux and what does not. Finally, we will be capable of making the decision whether we can already start working with Spring Boot 2 without fear.

Oleh Dokuka

May 21, 2018
Tweet

More Decks by Oleh Dokuka

Other Decks in Programming

Transcript

  1. 2

  2. 2

  3. 2

  4. 2

  5. 2

  6. 13

  7. 13

  8. 14

  9. 14

  10. !15

  11. !16

  12. !18

  13. !18

  14. !18

  15. Reactive Types abstract class Flux<T> implements Publisher<T> { ... }

    abstract class Flux<T> implements Publisher<T> { ... }
  16. interface WebSocketHandler { Mono<Void> handle(WebSocketSession session); } interface WebSocketHandler {

    Mono<Void> handle(WebSocketSession session); } interface WebSocketHandler { Mono<Void> handle(WebSocketSession session); } !31
  17. interface WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } interface

    WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } !32
  18. interface WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } interface

    WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } interface WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } !32
  19. interface WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } interface

    WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } interface WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } interface WebSocketSession { Flux<WebSocketMessage> receive(); Mono<Void> send(Publisher<WebSocketMessage> messages); } !32
  20. class WebSocketMessage { DataBuffer getPayload() String getPayloadAsText() WebSocketMessage retain() void

    release() } !33 class WebSocketMessage { DataBuffer getPayload() String getPayloadAsText() WebSocketMessage retain() void release() }
  21. class WebSocketMessage { DataBuffer getPayload() String getPayloadAsText() WebSocketMessage retain() void

    release() } !33 class WebSocketMessage { DataBuffer getPayload() String getPayloadAsText() WebSocketMessage retain() void release() } class WebSocketMessage { DataBuffer getPayload() String getPayloadAsText() WebSocketMessage retain() void release() }
  22. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); !39
  23. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); !39
  24. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !39
  25. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !40
  26. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !40
  27. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !41
  28. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !41
  29. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !42
  30. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !44
  31. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !46
  32. s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…))

    .map(db -> new WebSocketMessage(…)) .as(session::send); s.receive() .map(WebSocketMessage::retain) .map(WebSocketMessage::getPayload) .publishOn(Schedulers.parallel()) .transform(mapper::decode) .transform(this::doHandle) .onBackpressureBuffer() .transform(s -> mapper.encode(…)) .map(db -> new WebSocketMessage(…)) .as(session::send); .map(…) .publishOn(…) .onBackpressure() .tranform(…) .tranform(…) .map(…) !47
  33. interface WebSocketClient { Mono<Void> execute(URI url, WebSocketHandler handler); } interface

    WebSocketClient { Mono<Void> execute(URI url, WebSocketHandler handler); } !54
  34. interface WebSocketClient { Mono<Void> execute(URI url, WebSocketHandler handler); } interface

    WebSocketClient { Mono<Void> execute(URI url, WebSocketHandler handler); } !54 interface WebSocketClient { Mono<Void> execute(URI url, WebSocketHandler handler); }
  35. map(…) Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…)

    .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); publishOn(…) then() flatMapIterable(…) !58
  36. map(…) Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…)

    .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); publishOn(…) then() flatMapIterable(…) Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); !58
  37. map(…) Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…)

    .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); publishOn(…) then() flatMapIterable(…) Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); !58
  38. Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…)

    .doOnNext(sink::next) .then(); ... ); map(…) publishOn(…) then() flatMapIterable(…) !59
  39. Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…)

    .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); doOnNext(sink::next) map(…) publishOn(…) then() flatMapIterable(…) !59
  40. Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…)

    .doOnNext(sink::next) .then(); ... ); Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…) .doOnNext(sink::next) .then(); ... ); doOnNext(sink::next) map(…) publishOn(…) then() flatMapIterable(…) !59
  41. Flux.create(sink -> ... s -> s.receive() .skip(6) .map(WSM::getPayloadAsText) .publishOn(…) .flatMapIterable(…)

    .doOnNext(sink::next) .then(); ... ); doOnNext(sink::next) map(…) publishOn(…) then() flatMapIterable(…) !60
  42. 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(); } 75
  43. 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(); } 76
  44. Что нужно? •MongoDB - реактивная база данных •Spring Data Mongo

    Reactive •Конфигурация SpringSecurity - для пользовательского доступа !77
  45. ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName); ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscribe(out::println) !82

    {emptyMap} ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscribe(out::println)
  46. ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName); ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) ReactiveSecurityContextHolder

    .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) !83 {emptyMap}
  47. ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName); ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) ReactiveSecurityContextHolder

    .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) !83 {security}
  48. ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName); ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) ReactiveSecurityContextHolder

    .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) ReactiveSecurityContextHolder .getContext() .map(getAuthentication) .map(getName) .subscriberContext(security) .subscribe(out::println) !83 {security}
  49. !84

  50. Бизнес механизм •Снять деньги с кошелька •Провести торги •В случае

    успеха - добавить деньги в кошелек •В случае неудачных торгов - вернуть деньги •В случае нехватки средств - ничего не делать !92
  51. tradeOffer .onBackpressureBuffer() .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class,

    t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) ) .map(LocalMessageMapper::tradeToMessage) .doOnNext(stream.sink()::next) .then(); !102
  52. tradeOffer .onBackpressureBuffer() .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class,

    t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) ) .map(LocalMessageMapper::tradeToMessage) .doOnNext(stream.sink()::next) .then(); !102
  53. tradeOffer .onBackpressureBuffer() .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class,

    t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) ) .map(LocalMessageMapper::tradeToMessage) .doOnNext(stream.sink()::next) .then(); !102
  54. tradeOffer .onBackpressureBuffer() .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class,

    t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) ) .map(LocalMessageMapper::tradeToMessage) .doOnNext(stream.sink()::next) .then();!102
  55. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } !103
  56. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !103
  57. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !104
  58. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !104
  59. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !104
  60. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !104
  61. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !105
  62. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !105
  63. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !105
  64. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !107
  65. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !107
  66. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !107
  67. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !108
  68. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !108
  69. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !108
  70. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } !109
  71. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !109
  72. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !110
  73. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !110
  74. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(NEME) onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !112
  75. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(NEME) onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !112
  76. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } !113
  77. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !113
  78. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !114
  79. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !114
  80. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !114
  81. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !115
  82. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(rollback()) onErrorResume(NEME) adjust() doTrade() .withdraw() doStoreTrade() !115
  83. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(NEME) onErrorResume(rollback()) adjust() doTrade() .withdraw() doStoreTrade() onErrorResume(rollback()) !117
  84. .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t ->

    Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } .flatMap(trade -> ws.withdraw(trade) .then(doTrade(trade)) .then(ws.adjust(trade)) .then(doStoreTrade(trade)) .onErrorResume( NotEnoughMoneyException.class, t -> Mono.empty() ) .onErrorResume(t -> ws.rollback(trade) .then(Mono.empty()) ) } onErrorResume(NEME) onErrorResume(rollback()) adjust() doTrade() .withdraw() doStoreTrade() onErrorResume(rollback()) !117
  85. 130

  86. 130

  87. 130

  88. 130

  89. Что еще упустили? •Дебажить сложно - но просто тестить с

    StepVerifier •Нет реактивного JDBC (пока что) !142
  90. Что еще упустили? •Дебажить сложно - но просто тестить с

    StepVerifier •Нет реактивного JDBC (пока что) •Императивный код никто не отменял !142
  91. На посмотреть •WebFlux воркшоп - https://bclozel.github.io/webflux- workshop/ •Играем с Reactor

    3 - https://tech.io/playgrounds/929/ reactive-programming-with-reactor-3/Intro •JDBC? Смотреть тут - https://www.youtube.com/watch? v=OiXu05WU7zI •Все еще не уверены? - Netflix вещает - https://goo.gl/ dMBUC7 !144