Presentation: Reactive: Do. Or do not. There is no try.

1. Reactive DOs and DONTs Sergei Egorov, Pivotal @bsideup
2. • Staff Engineer at Pivotal, Project Reactor team • Oracle Groundbreakers Ambassador • Berlin Spring User Group co-organizer • Testcontainers co-maintainer About me @bsideup
3. My first experience with reactive programming was like…
4. @bsideup
5. @bsideup
6. 1 week later…
7. I agree with Heinrich Apfelmus that the essence of functional reactive programming is to specify the dynamic behavior of a value completely at the time of declaration. “Reactive Sir” @bsideup
8. 1 week of production later…
9. @bsideup
10. So I went to the Gods…
11. Stephane @smaldini Maldini, Project Reactor’s team lead @bsideup Ex-
12. He was my manager, lol Stephane @smaldini Maldini, Project Reactor’s team lead @bsideup Ex-
13. Stephane @smaldini Maldini, Project Reactor’s team lead @bsideup Ex-
14. The Reactive Lord has given unto you these fifteen…
15. oy…
16. Ten! Ten commandments, for all to obey.
17. Learn functional programming DO @bsideup
18. @bsideup Haskell?
19. @bsideup Haskell?
20. @bsideup https://twitter.com/mariofusco/status/571999216039542784
21. Lazy evaluation
22. Mono.fromRunnable() @bsideup
23. Mono.fromRunnable() My code @bsideup
24. Mono.fromRunnable() My code My code @bsideup
25. Mono.fromRunnable() My code My code My code Work! @bsideup
26. Mono.fromRunnable().subscribe() @bsideup
27. An ideal app subscribes only once @bsideup
28. Immutability
29. Quiz static <T> void addLogging(Flux<T> flux) { flux.doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... Flux<Item> items = getFlux(); addLogging(items); items.subscribe(); @bsideup
30. Quiz static <T> void addLogging(Flux<T> flux) { flux.doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... Flux<Item> items = getFlux(); addLogging(items); items.subscribe(); 1. Will print both items and errors @bsideup
31. Quiz static <T> void addLogging(Flux<T> flux) { flux.doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... Flux<Item> items = getFlux(); addLogging(items); items.subscribe(); 1. Will print both items and errors 2. Will only print items, not errors @bsideup
32. Quiz static <T> void addLogging(Flux<T> flux) { flux.doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... Flux<Item> items = getFlux(); addLogging(items); items.subscribe(); 1. Will print both items and errors 2. Will only print items, not errors 3. Will not print anything @bsideup
33. Quiz static <T> void addLogging(Flux<T> flux) { flux.doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... Flux<Item> items = getFlux(); addLogging(items); items.subscribe(); 1. Will print both items and errors 2. Will only print items, not errors 3. Will not print anything @bsideup
34. public <R> Mono<R> map(Function<? super T, ? extends R> mapper) { return new MonoMap<>(this, mapper); } Immutability of reactive operators @bsideup Mono.just("Hello") .map(it -> it + " World!") .map(String::length)
35. public <R> Mono<R> map(Function<? super T, ? extends R> mapper) { return new MonoMap<>(this, mapper); } Immutability of reactive operators @bsideup Mono.just("Hello") .map(it -> it + " World!") .map(String::length)
36. static <T> Flux<T> addLogging(Flux<T> flux) { return flux .doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... getFlux() .transform(flux -> addLogging(flux)) .subscribe(); @bsideup Always return…
37. static <T> Flux<T> addLogging(Flux<T> flux) { return flux .doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... getFlux() .transform(flux -> addLogging(flux)) .subscribe(); … and use the returned value! @bsideup
38. static <T> Flux<T> addLogging(Flux<T> flux) { return flux .doOnNext(it -> println("Received " + it)) .doOnError(e -> e.printStackTrace()); } // ... getFlux() .transform(flux -> addLogging(flux)) .subscribe(); … and use the returned value! @bsideup
39. Higher-order functions
40. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> System.out.println("User: " + user)); Higher-order functions @bsideup
41. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> System.out.println("User: " + user)); Higher-order functions @bsideup
42. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> System.out.println("User: " + user)); Higher-order functions @bsideup
43. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> System.out.println("User: " + user)); Higher-order functions @bsideup
44. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> println("User: " + user)); Higher-order functions @bsideup
45. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> println("User: " + user)); Higher-order functions Less scopes @bsideup
46. Mono<String> userIdMono = getUserId(); userIdMono.subscribe(userId -> { Mono<User> userMono = getUser(userId); userMono.subscribe(user -> { if (isUserValid(user)) { println("User: " + user); } }); }); // vs getUserId() .flatMap(userId -> getUser(userId)) .filter(user -> isUserValid(user)) .subscribe(user -> println("User: " + user)); Higher-order functions Composition @bsideup
47. Side effects are not welcomed
48. https:/ /xkcd.com/1312/ @bsideup
49. Avoid side effects Flux<User> users = getUsers(); return users.doOnNext(user -> { storeUser(user); }); @bsideup
50. Avoid side effects Flux<User> users = getUsers(); return users.doOnNext(user -> { storeUser(user); }); Side effect! void storeUser(User user) { // } @bsideup
51. Avoid side effects Flux<User> users = getUsers(); return users.concatMap(user -> storeUser(user)); Mono<Void> storeUser(User user) { // } @bsideup
52. “things you do in your (async) callbacks should never take (significant) time” Stephane Maldini, 2019 @bsideup
53. Errors are side effects too return fetchUsers().map(json -> { try { return decodeJSON(json); } catch (IOException e) { throw new RuntimeException(e); } }); @bsideup
54. Errors are side effects too return fetchUsers().map(json -> { try { return decodeJSON(json); } catch (IOException e) { throw new RuntimeException(e); } }); @bsideup
55. Errors are side effects too return fetchUsers().map(json -> { try { return decodeJSON(json); } catch (IOException e) { throw new RuntimeException(e); } }); return fetchUsers().handle((json, sink) -> { try { sink.next(decodeJSON(json)); } catch (IOException e) { sink.error(e); } }); @bsideup
56. Errors are side effects too return fetchUsers().map(json -> { try { return decodeJSON(json); } catch (IOException e) { throw new RuntimeException(e); } }); return fetchUsers().handle((json, sink) -> { try { sink.next(decodeJSON(json)); } catch (IOException e) { sink.error(e); } }); @bsideup
57. Errors are side effects too return fetchUsers().map(json -> { try { return decodeJSON(json); } catch (IOException e) { throw new RuntimeException(e); } }); return fetchUsers().handle((json, sink) -> { try { sink.next(decodeJSON(json)); } catch (IOException e) { sink.error(e); } }); No need to wrap with RuntimeException @bsideup
58. use it for heavy computations DON’T @bsideup
59. ~5x Overhead
60. .flatMap .concatMap(this::calculateHash) .delayUntil(this::squareRoot)
61. .flatMap .concatMap .delayUntil(this::squareRoot)
62. .flatMap .concatMap .publishOn
63. .flatMap .concatMap .publishOn Queue Queue Queue
66. check various operators DO @bsideup
67. “Reactive starter pack” • map • filter • flatMap • take • subscribeOn/publishOn @bsideup
68. @bsideup
69. @bsideup
70. “flatMaps” • flatMap - transform every item concurrently into a sub-stream, and join the current and the sub-stream. @bsideup
71. “flatMaps” • flatMap - transform every item concurrently into a sub-stream, and join the current and the sub-stream. • concatMap - same as flatMap, but one-by-one @bsideup
72. “flatMaps” • flatMap - transform every item concurrently into a sub-stream, and join the current and the sub-stream. • concatMap - same as flatMap, but one-by-one • switchMap - same as concatMap, but will cancel the previous sub- stream when a new item arrives @bsideup
73. “flatMaps” • flatMap - transform every item concurrently into a sub-stream, and join the current and the sub-stream. • concatMap - same as flatMap, but one-by-one • switchMap - same as concatMap, but will cancel the previous sub- stream when a new item arrives • flatMapSequential - same as flatMap, but preserves the order of sub- stream items according to the original stream’s order @bsideup
74. Which operator to use?
76. block non-blocking threads DON’T @bsideup
77. block non-blocking threads DON’T @bsideup
78. How reactive frameworks schedule tasks @bsideup
79. Default thread pools Schedulers.parallel() - N threads, where N matches the CPUs count. Schedulers.single() - 1 thread handling all submitted tasks Schedulers.boundedElastic() - dynamic, thread caching capped pool @bsideup
80. Mono.delay(ofSeconds(1)) @bsideup
81. Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) @bsideup
82. Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) Every non-instant operation 
 runs on the parallel scheduler
 by default! @bsideup
83. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
84. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
85. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
86. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
87. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
88. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
89. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
90. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
91. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
92. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
93. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
94. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
95. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
96. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
97. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution ⚠ no more tasks scheduled 
 until this task returns @bsideup
98. How to fix?
99. Project Loom
100. @bsideup
101. @bsideup
102. Use non-blocking APIs, or…
103. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
104. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
105. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
106. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
107. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
108. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
109. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
110. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
111. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
112. Demo http://eskipaper.com/homer-jay-simpson-cartoon.html
113. Demo outcomes • Blocking calls are bad, m’kay? • They may sneak into your production system • Use https://github.com/reactor/BlockHound to detect them • Supports multiple frameworks (Reactor, RxJava, etc) • … and maybe even Kotlin: 
 https://github.com/Kotlin/kotlinx.coroutines/issues/1031 • Use a dedicated pool for the necessary blocking calls, or schedule them on the Schedulers.boundedElastic() built-in pool if they happen rarely @bsideup
114. “I use async APIs and I am safe!”
115. Yeah… sure.
116. Or…
117. Are you sure?
118. KafkaProducer#send(ProducerRecord,Callback) @bsideup
119. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc @bsideup
120. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc java.lang.Error: Blocking call! java.lang.Object#wait at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154) at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254) at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43) at java.lang.Object.wait(Object.java) at org.apache.kafka.clients.Metadata.awaitUpdate(Metadata.java:181) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:938) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:823) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) @bsideup
121. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc java.lang.Error: Blocking call! java.lang.Object#wait at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154) at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254) at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43) at java.lang.Object.wait(Object.java) at org.apache.kafka.clients.Metadata.awaitUpdate(Metadata.java:181) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:938) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:823) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) https://issues.apache.org/jira/browse/KAFKA-3539 @bsideup
122. long remainingWaitMs = maxWaitMs; long elapsed; // Issue metadata requests until we have metadata for the topic or maxWaitTimeMs is exceeded. // In case we already have cached metadata for the topic, but the requested partition is greater // than expected, issue an update request only once. This is necessary in case the metadata // is stale and the number of partitions for this topic has increased in the meantime. do { log.trace("Requesting metadata update for topic {}.", topic); metadata.add(topic); int version = metadata.requestUpdate(); sender.wakeup(); try { metadata.awaitUpdate(version, remainingWaitMs); } catch (TimeoutException ex) { // Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms."); } cluster = metadata.fetch(); elapsed = time.milliseconds() - begin; if (elapsed >= maxWaitMs) throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms."); if (cluster.unauthorizedTopics().contains(topic)) throw new TopicAuthorizationException(topic); remainingWaitMs = maxWaitMs - elapsed; partitionsCount = cluster.partitionCountForTopic(topic); } while (partitionsCount == null); waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs); Default is “60 seconds” @bsideup
123. @bsideup
124. Start gradually DO @bsideup
125. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll(); } } interface UsersRepository { List<User> findAll(); } @bsideup
126. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll(); } } interface UsersRepository { List<User> findAll(); } @bsideup
127. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll(); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
128. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll(); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
129. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll() .collectList() .block(); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
130. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll() .collectList() .timeout(Mono.delay(ofSeconds(10))) .retry(5) .block(); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
131. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; Flux<User> getUsers() { return usersRepository.findAll() .timeout(Mono.delay(ofSeconds(10))) .retry(5); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
132. class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers().stream() .map(UserDTO::new) .collect(Collectors.toList()); } } class UsersService { UsersRepository usersRepository; Flux<User> getUsers() { return usersRepository.findAll() .timeout(Mono.delay(ofSeconds(10))) .retry(5); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
133. class UsersController { UsersService usersService; @GetMapping("/users") Flux<UserDTO> getUsers() { return usersService.getUsers() .map(UserDTO::new); } } class UsersService { UsersRepository usersRepository; Flux<User> getUsers() { return usersRepository.findAll() .timeout(Mono.delay(ofSeconds(10))) .retry(5); } } interface UsersRepository { Flux<User> findAll(); } @bsideup
134. Before: class UsersController { UsersService usersService; @GetMapping("/users") List<UserDTO> getUsers() { return usersService.getUsers() .stream() .map(UserDTO::new) .collect(toList()); } } class UsersService { UsersRepository usersRepository; List<User> getUsers() { return usersRepository.findAll(); } } interface UsersRepository { List<User> findAll(); } class UsersController { UsersService usersService; @GetMapping("/users") Flux<UserDTO> getUsers() { return usersService.getUsers() .map(UserDTO::new); } } class UsersService { UsersRepository usersRepository; Flux<User> getUsers() { return usersRepository.findAll() .timeout(Mono.delay(ofSeconds(10))) .retry(5); } } interface UsersRepository { Flux<User> findAll(); } After:
135. use ThreadLocals DON’T @bsideup
136. Synchronous programming • 1 request == 1 thread • Everything is blocking, the execution continues on the same thread • ThreadLocals work just fine @bsideup
137. Asynchronous programming • 1 request will most probably be handled by multiple threads • Everything is non-blocking, an accepted request may be returned to the client from another thread • ThreadLocals must be propagated from one thread to another @bsideup
138. Solution?
139. Context!
140. Mono .deferWithContext(ctx -> { return this.getUser(ctx.get("userId")); }) // Later in the framework (e.g. Spring Security) .subscriberContext(Context.of("userId", "bsideup")) .subscribe(); @bsideup
141. Mono .deferWithContext(ctx -> { return this.getUser(ctx.get("userId")); }) // Later in the framework (e.g. Spring Security) .subscriberContext(Context.of("userId", "bsideup")) .subscribe(); @bsideup
142. Mono .deferWithContext(ctx -> { return this.getUser(ctx.get("userId")); }) // Later in the framework (e.g. Spring Security) .subscriberContext(Context.of("userId", "bsideup")) .subscribe(); @bsideup
143. Legacy?
144. Schedulers.onScheduleHook(fn) @bsideup
145. Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("myHook", runnable -> { println("Before every scheduled runnable"); return () -> { println("Before execution"); runnable.run(); println("After execution"); }; });
146. Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("mdc", runnable -> { String userId = MDC.get("userId"); return () -> { MDC.put("userId", userId); try { runnable.run(); } finally { MDC.remove("userId"); } }; });
147. think about the resiliency DO @bsideup
148. “Error… is a signal.” Stephane Maldini, year unknown @bsideup
149. Resiliency with Reactor @bsideup
150. Resiliency with Reactor • .timeout(Duration) - cancel the subscription and fail if no items emitted @bsideup
151. Resiliency with Reactor • .timeout(Duration) - cancel the subscription and fail if no items emitted • .retry()/retryWithBackoff() - retry the subscription on failure @bsideup
152. Resiliency with Reactor • .timeout(Duration) - cancel the subscription and fail if no items emitted • .retry()/retryWithBackoff() - retry the subscription on failure • .repeatWhenEmpty() - repeat the subscription when it completes without values @bsideup
153. Resiliency with Reactor • .timeout(Duration) - cancel the subscription and fail if no items emitted • .retry()/retryWithBackoff() - retry the subscription on failure • .repeatWhenEmpty() - repeat the subscription when it completes without values • .defaultIfEmpty() - fallback when empty @bsideup
154. Resiliency with Reactor • .timeout(Duration) - cancel the subscription and fail if no items emitted • .retry()/retryWithBackoff() - retry the subscription on failure • .repeatWhenEmpty() - repeat the subscription when it completes without values • .defaultIfEmpty() - fallback when empty • .onErrorResume() - fallback on error @bsideup
155. Distributed?
156. https://github.com/resilience4j/resilience4j RateLimiter rateLimiter = RateLimiter.ofDefaults(“name"); Mono.fromCallable(backendService::doSomething) .transformDeferred(RateLimiterOperator.of(rateLimiter))
157. care about the threads DON’T @bsideup
158. You may ask… • “How do I get the current Scheduler?” • “Why does flatMap changes the thread?” • …”Why it does not?” @bsideup
159. The answer is…
160. You.
161. You. Should.
162. You. Should. Not.
163. You. Should. Not. Care.
164. Prepare for day 2 DO @bsideup
165. Demo http://eskipaper.com/homer-jay-simpson-cartoon.html
166. Demo outcomes • Use .checkpoint(“something”) to “mark” reactive “milestones” • Read about Hooks.onOperatorDebug()… • … but use reactor-tools’ ReactorDebugAgent (works in prod too) • https://spring.io/blog/2019/03/06/flight-of-the-flux-1-assembly-vs- subscription - great article from Simon Basle about the internals @bsideup
167. Be afraid of reactive programming ;) DON’T @bsideup
168. @bsideup
169. Spring Webflux @bsideup
170. Developer experience is one of the main focuses in Reactor 3.3…
171. … and we just started
172. Summary • Learn functional programming • Check various operators • Think about the resiliency • Start gradually • Prepare for day 2 • Use it for heavy computations • Care about the threads • Block non-blocking threads • Use ThreadLocals • Be afraid of it ;) DO… DON’T… @bsideup
173. @bsideup bsideup