Reactive System. What and How?

Transcript

1. Reactive System. What and How? by Oleh Dokuka
2. About me 2
3. • JSE at Levi9 About me 2
4. • JSE at Levi9 • JEEConf Speaker About me 2
5. • JSE at Levi9 • JEEConf Speaker • Spring fan About me 2
6. • JSE at Levi9 • JEEConf Speaker • Spring fan • Reactivity preacher About me 2
7. • JSE at Levi9 • JEEConf Speaker • Spring fan • Reactivity preacher /oleh.dokuka /OlegDokuka /OlehDokuka About me 2
8. 3 Agenda
9. 3 • Reactive Approach Agenda
10. 3 • Reactive System • Reactive Approach Agenda
11. 3 • Reactive System • Reactive Programming • Reactive Approach Agenda
12. 3 • Reactive System • Reactive Programming • Reactive Approach Agenda • Reactive In Action
13. Why Reactive? 4
17. Responsive? 8
18. 9 Responsive
19. 9 Responsive
20. 10 Responsive 10
21. 10 Responsive Elastic 10
22. 10 Responsive Elastic 10
23. Responsive Elastic
24. Responsive Elastic Resilient
25. Responsive Elastic Resilient
26. 12 Responsive Elastic Resilient 12
27. 12 Responsive Elastic Resilient 12
28. 13 Responsive Elastic Resilient
29. 13 Responsive Elastic Resilient Message Driven
30. 13 Responsive Elastic Resilient Message Driven
31. 13 Responsive Elastic Resilient Message Driven
32. 13 Responsive Elastic Resilient Message Driven
33. 14 Responsive Elastic Resilient Message Driven
34. Reactive Manifesto 15 http://www.reactivemanifesto.org/
35. Common Vocabulary 16
36. Reactive System 17
37. Architecture 18
38. 19 Message Driven
39. Reactive Programming 20
40. Reactive Programming != Reactive System 21
41. Reactive Programming Evolution 22
42. Observer Pattern Event Listening 23
49. new EventHandler() { void onEvent(Event event) { //do stuff } } 27
50. eventSource.setHandler( new EventHandler() { void onEvent(Event event) { //do stuff } } ); 28
51. anotherEventSource.setHandler( new EventHandler() { void onEvent(Event event) { eventSource.setHandler( new EventHandler() { void onEvent(Event event) { //do stuff } } ); } } ); 29
52. Pyramid of Doom 30
53. Async operation? 31
54. Framework’s Way 32
55. @Component EventHandler() { @EventListener void onEvent(Event event) { //do stuff } } 33
56. @Component EventHandler() { @EventListener void onEvent(Event event) { //do stuff } } 34
57. No Clear Flow 35
58. Reactive Extension 36
59. Rx.Net 37
60. 38 DataSource Subscriber
61. 38 DataSource Observable Subscriber
62. 39 Observable DataSource Subscriber
63. 39 Observable DataSource Observer Subscriber
64. 40 Observable DataSource Subscriber Observer
65. 40 Observable DataSource Subscriber onNext() onError() onComplete() Observer
66. 40 Observable DataSource Subscriber onNext() onError() onComplete() Observer
67. 41 onNext() onError() onComplete() Observable Observer Subscriber DataSource
68. 41 onNext() onError() onComplete() Observable Observer Subscriber DataSource Observer Transform onNext() onError() onComplete()
69. 41 onNext() onError() onComplete() Observable Observer Subscriber DataSource Observer Transform onNext() onError() onComplete()
70. Observer Transform 42 Observable Observer Subscriber DataSource onNext() onError() onComplete() onNext() onError() onComplete()
71. IObservable<T> 43
72. IObserver<T> 44
73. public interface IObserver<in T> { // Notifies the observer that // the source has finished sending messages. void OnCompleted(); // Notifies the observer about any // exception or error. void OnError(Exception error); // Pushes the next data value from // the source to the observer. void OnNext(T value); } 45
74. public interface IObserver<in T> { // Notifies the observer that // the source has finished sending messages. void OnCompleted(); // Notifies the observer about any // exception or error. void OnError(Exception error); // Pushes the next data value from // the source to the observer. void OnNext(T value); } 46
75. public interface IObserver<in T> { // Notifies the observer that // the source has finished sending messages. void OnCompleted(); // Notifies the observer about any // exception or error. void OnError(Exception error); // Pushes the next data value from // the source to the observer. void OnNext(T value); } 47
76. public interface IObserver<in T> { // Notifies the observer that // the source has finished sending messages. void OnCompleted(); // Notifies the observer about any // exception or error. void OnError(Exception error); // Pushes the next data value from // the source to the observer. void OnNext(T value); } 48
77. API Transformations, Operations, etc. 49
78. API .just() .from() .defer() .map() .filter() .flatMap() etc. 50
79. Benefit 51
80. observable.businessFunction() .map(mapFunction()) .filter(filterFunction()) .map(mapFunction()) .doOnNext(someAction()) .subscribe() 52
81. Clear flow! Looks imperative, but actually async non-blocking processing of events 53
82. Order Processing Flow 54
83. RxJava Successful Story of Copy/Paste 55
84. Observable<T> 56
85. class Observable<T> { ... static <T> Observable<T> from(T... items) static Observable<Integer> range(...) static <T> Observable<T> just(T value) ... } 57
86. class Observable<T> { ... <R> Observable<R> map(...) Observable<T> filter(...) <R> Observable<R> flatMap(...) ... } 58
87. Observer<T> 59
88. interface Observer<T> { void onCompleted(); void onError(Throwable e); void onNext(T args); } 60
89. Subscription 61
90. class Observable<T> { ... Subscription subscribe(Observer<T> observer){ ... } ... } 62
91. class Observable<T> { ... Subscription subscribe(Observer<T> observer){ ... } ... } 63
92. class Observable<T> { ... Subscription subscribe(Observer<T> observer){ ... } ... } 64
93. class Observable<T> { ... Subscription subscribe(Observer<T> observer){ ... } ... } 65
94. Flow Examples 66
95. Source Creation 67
96. Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) 68
97. Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) .just( ) a Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) 68
98. .just( ) a a Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) 69
99. .just( ) a a onNext() onError() onComplete() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) 69
100. .just( ) onNext() onError() onComplete() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) a 70
101. .just( ) .just( ) onNext() onNext() onError() onComplete() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) a 70
102. onNext() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) .just( ) a onError() onComplete() 71
103. onNext() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) onNext() .just( ) onError() onComplete() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) onComplete() 71
104. onComplete() onNext() .just( ) onError() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) 72
105. onComplete() onNext() .just( ) onError() onComplete() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) 72
106. Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) onNext() Observable .just("a") .subscribe( onNext(), onError(), onComplete() ) .just( ) onError() onComplete() onComplete() onComplete() 73
107. Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) 74
108. Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) .from(asList( )) a b 74
109. 75 Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) .from(asList( )) a b
110. 75 onNext() onError() onComplete() Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) .from(asList( )) a b
111. .from(asList( )) 76 onNext() onError() onComplete() a Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) b
112. .from(asList( )) .from(asList( )) 76 onNext() onError() onComplete() onNext() a Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) b
113. onNext() onComplete() .from(asList( )) onError() .from(asList( )) 77 onError() Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) a b
114. onComplete() onNext() onComplete() onNext() .from(asList( )) onError() .from(asList( )) 77 onError() Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() )
115. onNext() onComplete() onNext() .from(asList( )) onComplete() onError() .from(asList( )) 78 onError() Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() )
116. onNext() onComplete() onNext() .from(asList( )) onComplete() onError() .from(asList( )) 78 onError() onComplete() Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() ) Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() )
117. .from(asList( )) onComplete() onError() 79 onNext() Observable .from(asList("a","b")) .subscribe( onNext(), onError(), onComplete() )
118. 80 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
119. 80 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .defer() Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
120. Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) 81 .defer()
121. Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) 81 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .defer() onNext() onError() onComplete()
122. Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) onNext() onError() onComplete() 82 .defer()
123. Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) onNext() onError() onComplete() 82 .just( ) a b .defer() .defer()
124. .just( ) onNext() onError() onComplete() 83 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) a b .defer()
125. .just( ) onNext() onNext() onError() onComplete() 83 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) a b .defer() Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
126. onNext() onError() onComplete() 84 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) .defer() a b
127. onNext() onNext() onError() onComplete() onComplete() 84 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) .defer() Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
128. onComplete() onNext() onError() 85 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) .defer()
129. onComplete() onNext() onError() onComplete() 85 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) .defer() Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
130. onNext() onError() onComplete() onComplete() onComplete() 86 Observable .defer(() -> just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) .defer()
131. Transformations 87
132. 88 .map(…) .map()
133. 88 .map(…) .map()
134. 89 .filter(…) .filter()
135. 89 .filter(…) .filter()
136. 90 .flatMap(…) .flatMap()
137. 90 .flatMap(…) .flatMap()
138. externalService .messagesStream() .map(...) .filter(...) 91
139. externalService .messagesStream() .map(...) .filter(...) 91 .map() .filter() externalService .messagesStream() .map(...) .filter(...) .subscribe()
140. externalService .messagesStream() .map(...) .filter(...) .subscribe() .filter() .map() 92
141. externalService .messagesStream() .map(...) .filter(...) .subscribe() externalService .messagesStream() .map(...) .filter(...) .subscribe() .filter() .map() 92
142. externalService .messagesStream() .map(...) .filter(...) .subscribe() .map() .filter() 93
143. externalService .messagesStream() .map(...) .filter(...) .subscribe() externalService .messagesStream() .map(...) .filter(...) .subscribe() .map() .filter() 93
144. externalService .messagesStream() .map(...) .filter(...) .subscribe() .map() 94 .filter()
145. externalService .messagesStream() .map(...) .filter(...) .subscribe() externalService .messagesStream() .map(...) .filter(...) .subscribe() .map() 94 .filter()
146. externalService .messagesStream() .map(...) .filter(...) .subscribe() .map() .filter() 95
147. Thread Manipulation 96
148. Thread A Thread Main .observeOn(…) .observeOn() 97
149. Thread A Thread Main .observeOn(…) .observeOn() 97
150. 98 .subscribeOn() Thread Main Thread A .subscribeOn(…)
151. 98 .subscribeOn() Thread Main Thread A .subscribeOn(…)
152. 98 .subscribeOn() Thread Main Thread A .subscribeOn(…)
153. Quiz 99
154. Observable .just(1, 20, 300, 4000) .map(Object::toString) .filter(v -> v.length() > 3) 100 Output?
155. Observable .just(1, 20, 300, 4000) .map(Object::toString) .filter(v -> v.length() > 3) .subscribe(System.out::println) Output? 101
156. Output? 102 Observable .just(1, 20, 300, 4000) .map(Object::toString) .filter(v -> v.length() > 3) .subscribe(System.out::println)
157. Hot vs. Cold Observables 103
158. Cold 104
159. .from( ) 4 3 2 1 Observable .just(1, 2, 3, 4) 1 2 3 4 105
160. .from( ) 4 3 2 1 Observable .just(1, 2, 3, 4) 1 2 3 4 .subscribe() 105
161. .from( ) 4 3 2 1 Observable .just(1, 2, 3, 4) 1 2 3 4 1 2 3 4 .subscribe() 106
162. .from( ) 4 3 2 1 Observable .just(1, 2, 3, 4) 1 2 3 4 1 2 3 4 .subscribe() .subscribe() 106
163. .from( ) 4 3 2 1 Observable .just(1, 2, 3, 4) 1 2 3 4 1 2 3 4 .subscribe() .subscribe() 107
164. Hot 108
165. externalService .messagesStream() .map(...) .filter(...) 109
166. externalService .messagesStream() .map(...) .filter(...) externalService .messagesStream() .map(...) .filter(...) .map() .filter() 109
167. externalService .messagesStream() .map(...) .filter(...) .filter() .map() 110
168. externalService .messagesStream() .map(...) .filter(...) externalService .messagesStream() .map(...) .filter(...) .filter() .map() 110
169. externalService .messagesStream() .map(...) .filter(...) .map() .filter() 111
170. externalService .messagesStream() .map(...) .filter(...) externalService .messagesStream() .map(...) .filter(...) .map() .filter() 111
171. externalService .messagesStream() .map(...) .filter(...) .map() .filter() 112
172. externalService .messagesStream() .map(...) .filter(...) .map() externalService .messagesStream() .map(...) .filter(...) .filter() 112
173. externalService .messagesStream() .map(...) .filter(...) .map() .filter() 113
174. externalService .messagesStream() .map(...) .filter(...) .map() .filter() 114
175. .subscribe() externalService .messagesStream() .map(...) .filter(...) .map() .filter() 114
176. .subscribe() externalService .messagesStream() .map(...) .filter(...) .map() .filter() 115
177. RxJava Cons 116
178. RxJava Cons 116
179. RxJava Cons • Not easy to understand • Unexpected behaviors • Complicated extensibility • No backpressure 117
181. Problem 119
182. Reactive Streams 120
183. Purpose? 121
184. Common API 122
185. 123 Publisher<T>
186. 123 Publisher<T> Observable<T> like
187. 124 Subscriber<T>
188. 124 Subscriber<T> Observer<T> like
189. 125 public interface Publisher<T> { public void subscribe( Subscriber<? super T> s ); }
190. 126 .publish( ) a Publisher<String>
191. 126 .publish( ) a Subscriber<String> .subscribe() Publisher<String> Publisher<String>
192. 127 Subscription
193. 128 public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); }
194. 128 public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); } public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); }
195. public interface Subscription { public void request(long n); public void cancel(); } 129
196. public interface Subscription { public void request(long n); public void cancel(); } 129 public interface Subscription { public void request(long n); public void cancel(); }
197. Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } 130 .publish( ) a .subscribe() Publisher<String>
198. Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } 130 .publish( ) a .subscribe() Publisher<String>
199. Publisher<String> Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } 131 .publish( ) .subscribe() a
200. Publisher<String> Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } 131 .publish( ) .subscribe() Publisher<String> a
201. Publisher<String> Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } Subscriber<String> { void onSubscribe(Subscription s){ s.request(1); } … } 131 .publish( ) .subscribe() Publisher<String> a
202. Backpressure Support 132
203. 133 Processor<T, R>
204. 134 interface Processor<T, R> extends Subscriber<T>, Publisher<R> { }
205. 135 .publish( ) .subscribe() Publisher<String> a Subscriber<String>
206. 135 .publish( ) .subscribe() Publisher<String> a .process() Processor<String, String> Subscriber<String> Publisher<String> Subscriber<String>
207. Processor<String, String> .publish( ) 136 .subscribe() Publisher<String> a .process() Subscriber<String>
208. Processor<String, String> Processor<String, String> { s.request(1); } Processor<String, String> { s.request(1); } .publish( ) 136 .subscribe() Publisher<String> a .process() Subscriber<String>
209. Processor<String, String> Processor<String, String> { s.request(1); } .publish( ) b .process() 137 .subscribe() Publisher<String> a Subscriber<String>
210. Processor<String, String> Processor<String, String> { s.request(1); } .publish( ) b .process() 137 .subscribe() Publisher<String> Subscriber<String> a
211. Processor<String, String> { s.request(1); } .publish( ) b .process() 138 .subscribe() Publisher<String> Subscriber<String> a
212. Processor<String, String> { s.request(1); } Processor<String, String> { s.request(1); } Processor<String, String> { s.request(1); } .publish( ) b .process() 138 .subscribe() Publisher<String> Subscriber<String> a
213. Processor<String, String> Processor<String, String> { s.request(1); } .publish( ) .process() 139 .subscribe() Publisher<String> Subscriber<String> a b
214. Processor<String, String> Processor<String, String> { s.request(1); } .publish( ) .process() 139 .subscribe() Publisher<String> Subscriber<String> a b
215. Subscriber<String> Processor<String, String> 140 .publish( ) .subscribe() Publisher<String> .process() Processor<String, String> b a
216. Subscriber<String> Subscriber<String> { s.request(2); } Processor<String, String> 140 .publish( ) .subscribe() Publisher<String> .process() Processor<String, String> b a
217. Subscriber<String> Processor<String, String> 141 .publish( ) .subscribe() Publisher<String> .process() Processor<String, String> Subscriber<String> b a
218. Subscriber<String> Processor<String, String> 141 .publish( ) .subscribe() Publisher<String> .process() Processor<String, String> Subscriber<String> b a Subscriber<String>
219. Subscriber<String> Subscriber<String> Processor<String, String> 142 .publish( ) .subscribe() Publisher<String> .process() Processor<String, String> b a
220. Project Reactor 143
221. by 144
222. RxJava-like API 145
223. Benefit 146
224. Benefit 146
225. Reactive Types 147
226. Flux<T> 148
227. Flux<T> 148 Observable<T> like
228. 149 abstract class Flux<T> implements Publisher<T> { ... static <T> Flux<T> from(...) static Flux<Integer> range(...) static <T> Flux<T> just(T data) ... }
229. 150 abstract class Flux<T> implements Publisher<T> { ... <R> Flux<R> map(...) Flux<T> filter(...) <R> Flux<R> flatMap(...) ... }
230. Mono<T> 151
231. Mono<T> 151 ????????????? like
233. Mono<T> 153 Observable<T> like
234. BUT! 154
235. For Zero or One element 155
236. 156 Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .just( ) a b
237. 157 Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .just( ) a b
238. 157 Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .just( ) a b onNext() onError() onComplete() Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
239. .from( ____ ) .just( ) a b Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) 158 onNext() onError() onComplete()
240. .from( ____ ) .just( ) a b Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) 158 Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .just( ) a b onNext() onError() onComplete()
241. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .just( ) a b 159 .from( ____ ) .just( ) b onNext() onError() onComplete() a
242. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .just( ) a b Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) 159 .from( ____ ) .just( ) b onNext() onError() onComplete() onNext() a
243. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .from( ____ ) .just( ) a b 160 .just( ) onNext() onError() onComplete() a b
244. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .from( ____ ) .from( ____ ) .just( ) a b 160 .just( ) onNext() onError() onComplete() a Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( ) b
245. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .from( ____ ) .from( ____ ) .just( ) a b 160 .just( ) onNext() onError() onComplete() a Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .just( )
246. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .from( ____ ) .just( ) a b 161 .just( ) onNext() onError() onComplete() a .just( )
247. Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() ) .from( ____ ) .from( ____ ) .just( ) a b 161 .just( ) onNext() onError() onComplete() a .just( ) onComplete() Mono.from( Flux.just("a", "b") ) .subscribe( onNext(), onError(), onComplete() )
248. API Operations, Transformations, etc. 162
249. API .merge() & .mergeWith() .zip() & .zipWith() .publish() etc. 163
250. .merge() .merge(…) 164
251. .merge() .merge(…) 164
252. 165 Flux.merge( Flux.just("a"), Flux.just("b") ) .mergeWith(Flux.just("c"));
253. .zip() .zip(…) 166
254. .zip() .zip(…) 166
255. .zip() .zip(…) 166
256. .zip(…) 167 .zip()
257. .zip(…) 167 .zip()
258. 168 Flux.zip( Flux.just("a"), Flux.just("b"), (a, b) -> a + b ) .zipWith(Flux.just("c"), (ab, c) -> ab + c);
259. 169 .from()
260. 169 .from() static <T> Flux<T> from(Publisher<T> source) static <T> Mono<T> from(Publisher<T> source)
261. 170 Flux .from(externalPublisher()) .from()
262. 171 Mono .from(externalPublisher()) .from()
263. 172 .fromStream()
264. 172 .fromStream() static <T> Flux<T> fromStream(Stream<T> s)
265. 173 Flux.fromStream( Stream.of("a", "b") )
266. Backpressure Support 174 .onBackpressureBuffer() .onBackpressureDrop() .onBackpressureLatest() etc.
267. Summary 1st Part 175
268. Reactivity in Action 176
269. Action Plan Refactoring 177
270. Chat Application 178
271. Functionality • Gitter Chat Communication • Messaging Statistics 179
272. https://goo.gl/XcnEnr 180
273. Demo 181
274. Toolkit • Spring Framework 4.x • Spring MVC • Spring Data JPA • Spring Boot 1.5.x 182
275. Architecture 183
277. User Client 184
278. Gitter Service 184
279. Whole Application 184
280. Controllers Layer 184
281. Service Layer 184
282. DB Layer 184
283. Controllers 185
284. Thymeleaf View 186
285. REST Resources 187
286. Services 188
287. External Chats SPI 189
288. Domain Services 190
289. Gitter Integration 191
290. Service Implementations 192
291. Persistence 193
292. Database Model 194
293. Existing Flow 195
302. Problems 200
303. Problems • Pulling model 201
306. Problems • Pulling model • Blocking I/O 203
307. Solutions 204
308. Reactive Manifesto 205
312. Async Messaging Push Model 207
313. Gitter Streaming API 208
319. Async Messaging Elasticity 211
320. Reactive DB Driver 212
326. Async Messaging Push Model 215
327. WebSocket 216
330. One Solution for Everything 218
331. Spring 5 :) 219
332. Painless Replacement 220
333. Reactive Stack (WebFlux) @Controller, @RequestMapping,… Spring MVC Servlet API Servlet Container Spring Web Reactive Reactive HTTP Servlet, Netty, Undertow 221
334. @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(); } } 222
335. @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(); } } 223
336. @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()); } 224
337. @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()); } 225
341. Why Netty? 228
342. Asynchronous event-driven framework 229
343. New Reactive Architecture 230
344. Data Base Connection Pipe Gitter Connection Pipe 231 Client Connection Pipe
345. Data Base Connection Pipe Gitter Connection Pipe 231 Client Connection Pipe
346. Refactoring Steps 232
347. 1. Update to Spring 5 2. Reactive Interfaces 3. Reactive Services 4. Reactive Endpoints 233
348. https://goo.gl/yxAJGI 234
349. Update To Spring 5 235
350. Code Session 236
351. Reactive Interfaces 237
352. Reactive DB Access ReactiveCrudRepository<T, ID> 238
353. 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(); } 239
354. 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(); } 240
355. Code Session 241
356. Reactive Services 242
357. Gitter Service 243
358. Reactive WebClient Out-of-the-box 244
359. 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 } } 245
360. 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 } } 246
363. Code Session 248
364. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } 249
365. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } 249 .getMS() .getLM() .retry() cache
366. 250 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache
367. 250 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache
368. .getMS() .getLM() .retry() cache 251 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
369. .getMS() .getLM() .retry() cache 251 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
370. .getLM() .getMS() .retry() cache 252 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
371. .getLM() .getMS() .retry() cache 252 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
372. .getMS() .getLM() .retry() cache 253 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
373. .getMS() .getLM() .retry() cache 253 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getLM()
374. .getMS() .getLM() .retry() cache 254 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
375. .getMS() .getLM() .retry() cache 254 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
376. .getMS() .getLM() .retry() cache 255 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
377. .getMS() .getLM() .retry() cache 255 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
378. .getMS() .getLM() .retry() cache 256 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
379. .getMS() .getLM() .retry() cache 256 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
380. .getMS() .getLM() .retry() cache 257 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
381. .getMS() .getLM() .retry() cache 257 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
382. 258 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache
383. …subscribe(); 258 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
384. …subscribe(); 258 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); }
385. …subscribe(); 259 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache
386. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } …subscribe(); 260 .getMS() .getLM() .retry() cache
387. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } …subscribe(); 260 .getMS() .getLM() .retry() cache .getMS() .retry()
388. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } …subscribe(); 261 .getMS() .getLM() .retry() cache
389. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } …subscribe(); 261 .getMS() .getLM() .retry() cache …subscribe();
390. @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } …subscribe(); 261 .getMS() .getLM() .retry() cache …subscribe();
391. …subscribe(); 262 @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } .getMS() .getLM() .retry() cache @Autowired public GitterService( GitterClient gitterClient ) { gas = Flux.merge( getLatestMessages(), getMessagesStream(null) ) .retryWhen(...) .replay(30) .autoConnect(0); } …subscribe();
392. Message Service 263
395. Code Session 265
396. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, ApplicationContext context ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } 266
397. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 267
398. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 268
399. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 268
400. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 269
401. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 269
402. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 270
403. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 271
404. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 271
405. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 272
406. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 272
407. @Autowired public DefaultMessageService( MessageRepository messageRepository, ChatService<MessageResponse> chatClient, MessageBroker messageBroker ) { this.chatClient = chatClient; chatClient.stream() .transform(MessageMapper::toDomainUnits) .transform(messageRepository::saveAll) .subscribe() messageBroker.createChannel( "statisticChanged", saved.materialize() ); } .transform() 273
408. @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() 274
409. @Override public Flux<MessageVM> latest() { return chatClient.stream() .transform(MessageMapper ::toViewModelUnits ); } .transform() 274
410. Statistics Service 275
413. Code Session 277
414. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 278
415. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 279
416. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 279
417. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 280
418. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 280
419. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 281
420. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 282
421. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 282
422. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 283
423. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 283
424. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 284
425. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 285
426. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 285
427. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 286
428. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 286
429. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 287
430. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 287
431. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 288
432. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 289
433. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 289
434. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 290
435. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 290
436. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 291
437. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 291
438. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 292
439. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) null 292
440. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .map(…) .defaultIfEmpty(…) .zip(…) 293
441. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 293
442. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 294
443. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 295
444. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 295
445. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 296
446. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 296
447. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 297
448. private Mono<UsersStatisticVM> doGetUserStatistic Mono<UserVM> topActiveUser = userRepository .findMostActive() .map(UserMapper::toViewModelUnits); Mono<UserVM> topMentionedUser = userRepository .findMostPopular() .map(UserMapper::toViewModelUnits); return Mono.zip( topActiveUser.defaultIfEmpty(EMPTY_USER), topMentionedUser.defaultIfEmpty(EMPTY_USER), UsersStatisticVM::new ); } .defaultIfEmpty(…) .zip(…) .subscribe() 297
449. Reactive Endpoints 298
452. Reactive WebSocket Out-of-the-box 300
453. 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); } 301
454. 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); } 302
455. 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); } 303
456. Code Session 304
457. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 305
458. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 306
459. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 306
460. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 307
461. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 308
462. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 308
463. .merge() @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network 309
464. .merge() @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network 309
465. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 310
466. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 310
467. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 311
468. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 312
469. @Override public Publisher<?> handle( Publisher<Object> in ) { return Flux.merge( messageService.latest(), statisticService .usersStatisticStream() ); } network .merge() 312
470. New Reactive Architecture 313
473. Demo 315
474. https://goo.gl/XcnEnr 316
475. Checklist • Message-Driven • Resilient • Responsive • Elastic 317
476. Checklist • Message-Driven • Resilient • Responsive • Elastic 317
477. Checklist • Message-Driven • Resilient • Responsive • Elastic 317
478. Checklist • Message-Driven • Resilient • Responsive • Elastic 317
479. Checklist • Message-Driven • Resilient • Responsive • Elastic 317
480. Summary 318
481. Q&A 319
482. 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] 320
483. Thank you! 321