Connect.Tech 2017: Dive into RxJS Observables

Transcript

1. Dive into RxJS Jeremy Fairbank @elpapapollo / jfairbank Observables
2. Software is broken. We are here to fix it. Say hi@testdouble.com
3. Async
4. Async API Callback
5. Callback fetchUserById(1, function(err, user) { if (err) { console.error('Could not retrieve user'); } else { console.log(user); } });
6. }); }); }); }); }); }); Callback Callback Callback Callback Callback Callback
7. function fetchCustomerNameForOrder(orderId, done, fail) { fetchOrder(orderId, function(err, order) { if (err) { logError(err); fail(err); } else { fetchCustomer( order.customerId, function(err, customer) { if (err) { logError(err); fail(err); } else { done(customer.name); } } ); } }); }
8. function fetchCustomerNameForOrder(orderId, done, fail) { fetchOrder(orderId, function(err, order) { if (err) { logError(err); fail(err); } else { fetchCustomer( order.customerId, function(err, customer) { if (err) { logError(err); fail(err); } else { done(customer.name); } } ); } }); }
9. function fetchCustomerNameForOrder(orderId, done, fail) { fetchOrder(orderId, function(err, order) { if (err) { logError(err); fail(err); } else { fetchCustomer( order.customerId, function(err, customer) { if (err) { logError(err); fail(err); } else { done(customer.name); } } ); } }); }
10. Async API ?
11. Promise function fetchCustomerNameForOrder(orderId) { return fetchOrder(orderId) .then(order => fetchCustomer(order.customerId)) .then(customer => customer.name); }
12. .then(...) .then(...) .then(...) .then(...) .then(...) .then(...) .then(...) .then(...)
13. Error Error Error Error
14. function fetchCustomerNameForOrder(orderId) { return fetchOrder(orderId) .then(order => fetchCustomer(order.customerId)) .then(customer => customer.name) .catch(err => { logError(err); throw err; }); }
15. Getting there…
16. • More readable and maintainable async code • Better error handling • More declarative and versatile syntax • Capable of handling events, streams, and HTTP
17. RxJS Observables
18. Arrays [ 1, 2, 3, 4, 5 ] Sequences in space
19. Observables Sequences in time 1 2 3 4 5
20. Reactive
21. Reactive 3
22. const { Observable } = require('rxjs'); const source = Observable.of(1, 2, 3); source.subscribe(x => console.log(x)); // 1 // 2 // 3
23. const { Observable } = require('rxjs'); const source = Observable.of(1, 2, 3); source.subscribe(x => console.log(x)); // 1 // 2 // 3
24. const { Observable } = require('rxjs'); const source = Observable.of(1, 2, 3); source.subscribe(x => console.log(x)); // 1 // 2 // 3
25. const { Observable } = require('rxjs'); const source = Observable.of(1, 2, 3); source.subscribe(x => console.log(x)); // 1 // 2 // 3
26. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe
27. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe 1
28. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe 1
29. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe
30. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe 2
31. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe 2
32. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe
33. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe 3
34. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); console.log subscribe 3
35. Declarative Transformation Operate on events
36. Observable.of(1, 2, 3) .map(n => n * 2) .subscribe(x => console.log(x)); // 2 // 4 // 6
37. Observable.of(1, 2, 3) .map(n => n * 2) .subscribe(x => console.log(x)); // 2 // 4 // 6
38. Observable.of(1, 2, 3) console.log n * 2 map subscribe
39. Observable.of(1, 2, 3) console.log n * 2 map subscribe 1
40. Observable.of(1, 2, 3) console.log n * 2 map subscribe 1
41. Observable.of(1, 2, 3) console.log n * 2 map subscribe 2
42. Observable.of(1, 2, 3) console.log n * 2 map subscribe 2
43. Observable.of(1, 2, 3) console.log n * 2 map subscribe
44. Observable.of(1, 2, 3) console.log n * 2 map subscribe 2
45. Observable.of(1, 2, 3) console.log n * 2 map subscribe 2
46. Observable.of(1, 2, 3) console.log n * 2 map subscribe 4
47. Observable.of(1, 2, 3) console.log n * 2 map subscribe 4
48. Observable.of(1, 2, 3) console.log n * 2 map subscribe
49. Observable.of(1, 2, 3) console.log n * 2 map subscribe 3
50. Observable.of(1, 2, 3) console.log n * 2 map subscribe 3
51. Observable.of(1, 2, 3) console.log n * 2 map subscribe 6
52. Observable.of(1, 2, 3) console.log n * 2 map subscribe 6
53. Lazy Transformation Do only as much work as needed
54. Observable.range(1, 100) .map(n => n * 2) .filter(n => n > 4) .take(2) .subscribe(x => console.log(x)); // 6 // 8
55. Observable.range(1, 100) .map(n => n * 2) .filter(n => n > 4) .take(2) .subscribe(x => console.log(x)); // 6 // 8
56. Observable.range(1, 100) .map(n => n * 2) .filter(n => n > 4) .take(2) .subscribe(x => console.log(x)); // 6 // 8
57. Observable.range(1, 100) .map(n => n * 2) .filter(n => n > 4) .take(2) .subscribe(x => console.log(x)); // 6 // 8
58. Observable.range(1, 100) .map(n => n * 2) .filter(n => n > 4) .take(2) .subscribe(x => console.log(x)); // 6 // 8
59. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take
60. console.log n * 2 map subscribe n > 4 1 Observable.range(1, 100) filter 2 take
61. console.log n * 2 map subscribe n > 4 1 Observable.range(1, 100) filter 2 take
62. console.log n * 2 map subscribe n > 4 2 Observable.range(1, 100) filter 2 take
63. console.log n * 2 map subscribe n > 4 2 Observable.range(1, 100) filter 2 take
64. console.log n * 2 map subscribe n > 4 2 Observable.range(1, 100) filter 2 take ×
65. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take
66. console.log n * 2 map subscribe n > 4 2 Observable.range(1, 100) filter 2 take
67. console.log n * 2 map subscribe n > 4 2 Observable.range(1, 100) filter 2 take
68. console.log n * 2 map subscribe n > 4 4 Observable.range(1, 100) filter 2 take
69. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take 4
70. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take 4 ×
71. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take
72. console.log n * 2 map subscribe n > 4 3 Observable.range(1, 100) filter 2 take
73. console.log n * 2 map subscribe n > 4 3 Observable.range(1, 100) filter 2 take
74. console.log n * 2 map subscribe n > 4 6 Observable.range(1, 100) filter 2 take
75. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take 6
76. ✓ console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take 6
77. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 2 take 6
78. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 1 take 6
79. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 1 take
80. console.log n * 2 map subscribe n > 4 4 Observable.range(1, 100) filter 1 take
81. console.log n * 2 map subscribe n > 4 4 Observable.range(1, 100) filter 1 take
82. console.log n * 2 map subscribe n > 4 8 Observable.range(1, 100) filter 1 take
83. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 1 take 8
84. ✓ console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 1 take 8
85. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 1 take 8
86. console.log n * 2 map subscribe n > 4 Observable.range(1, 100) filter 0 take 8
87. DOM Events
88. let counter = 0; function updateCounter(n) { counter += n; counterEl.innerHTML = counter; } incrementBtn.addEventListener('click', () => { updateCounter(1); }); decrementBtn.addEventListener('click', () => { updateCounter(-1); });
89. let counter = 0; function updateCounter(n) { counter += n; counterEl.innerHTML = counter; } incrementBtn.addEventListener('click', () => { updateCounter(1); }); decrementBtn.addEventListener('click', () => { updateCounter(-1); });
90. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
91. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
92. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
93. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
94. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
95. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
96. Observable.fromEvent(incrementBtn, 'click') .mapTo(1) .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
97. .scan((acc, curr) => acc + curr, 0) Accumulated counter value
98. .scan((acc, curr) => acc + curr, 0) Current event value (1)
99. .scan((acc, curr) => acc + curr, 0) Return new accumulated counter value
100. .scan((acc, curr) => acc + curr, 0) Initial counter value
101. counterEl.innerHTML = counter acc + curr 1 mapTo subscribe 0 scan
102. acc + curr 1 mapTo subscribe e scan 0 counterEl.innerHTML = counter
103. 1 mapTo subscribe e scan acc + curr 0 counterEl.innerHTML = counter
104. 1 mapTo subscribe 1 scan acc + curr 0 counterEl.innerHTML = counter
105. 1 mapTo subscribe scan 1 acc + curr 0 counterEl.innerHTML = counter
106. 1 mapTo subscribe scan 1 acc + curr 1 Counter value counterEl.innerHTML = counter
107. 1 mapTo subscribe scan 1 acc + curr 1 counterEl.innerHTML = counter
108. acc + curr 1 mapTo subscribe scan 1 counterEl.innerHTML = counter
109. acc + curr 1 mapTo subscribe e scan 1 counterEl.innerHTML = counter
110. 1 mapTo subscribe e scan acc + curr 1 counterEl.innerHTML = counter
111. 1 mapTo subscribe 1 scan acc + curr 1 counterEl.innerHTML = counter
112. 1 mapTo subscribe scan 1 acc + curr 1 counterEl.innerHTML = counter
113. 1 mapTo subscribe scan 2 acc + curr 2 Counter value counterEl.innerHTML = counter
114. 1 mapTo subscribe scan 2 acc + curr 2 counterEl.innerHTML = counter
115. const source = Observable.merge( Observable.fromEvent(incrementBtn, 'click').mapTo(1), Observable.fromEvent(decrementBtn, 'click').mapTo(-1), ); source .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
116. const source = Observable.merge( Observable.fromEvent(incrementBtn, 'click').mapTo(1), Observable.fromEvent(decrementBtn, 'click').mapTo(-1), ); source .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
117. const source = Observable.merge( Observable.fromEvent(incrementBtn, 'click').mapTo(1), Observable.fromEvent(decrementBtn, 'click').mapTo(-1), ); source .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
118. const source = Observable.merge( Observable.fromEvent(incrementBtn, 'click').mapTo(1), Observable.fromEvent(decrementBtn, 'click').mapTo(-1), ); source .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
119. const source = Observable.merge( Observable.fromEvent(incrementBtn, 'click').mapTo(1), Observable.fromEvent(decrementBtn, 'click').mapTo(-1), ); source .scan((acc, curr) => acc + curr, 0) .subscribe((counter) => { counterEl.innerHTML = counter; });
120. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement counterEl.innerHTML = counter
121. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement e counterEl.innerHTML = counter
122. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement e counterEl.innerHTML = counter
123. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement 1 counterEl.innerHTML = counter
124. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement 1 counterEl.innerHTML = counter
125. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement 1 counterEl.innerHTML = counter
126. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement 1 counterEl.innerHTML = counter
127. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement counterEl.innerHTML = counter
128. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement e counterEl.innerHTML = counter
129. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement e counterEl.innerHTML = counter
130. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement -1 counterEl.innerHTML = counter
131. acc + curr 1 mapTo subscribe 1 scan -1 mapTo Increment Decrement -1 counterEl.innerHTML = counter
132. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement 0 counterEl.innerHTML = counter
133. acc + curr 1 mapTo subscribe 0 scan -1 mapTo Increment Decrement 0 counterEl.innerHTML = counter
134. Async HTTP
135. Promise fetchOrders() .then((orders) => { orders.forEach((order) => { console.log(order); }); });
136. Promise fetchOrders() .then((orders) => { orders.forEach((order) => { console.log(order); }); });
137. Observable fetchOrders() .subscribe((orders) => { orders.forEach((order) => { console.log(order); }); });
138. Why Observables?
139. Promise fetchOrders() .then(orders => orders.filter( order => order.customerName === 'Tucker' )) .then(orders => orders.map(order => order.id)) .then((orderIds) => { orderIds.forEach(id => console.log(id)); });
140. Promise fetchOrders() .then(orders => orders.filter( order => order.customerName === 'Tucker' )) .then(orders => orders.map(order => order.id)) .then((orderIds) => { orderIds.forEach(id => console.log(id)); });
141. Promise fetchOrders() .then(orders => orders.filter( order => order.customerName === 'Tucker' )) .then(orders => orders.map(order => order.id)) .then((orderIds) => { orderIds.forEach(id => console.log(id)); });
142. Promise fetchOrders() .then(orders => orders.filter( order => order.customerName === 'Tucker' )) .then(orders => orders.map(order => order.id)) .then((orderIds) => { orderIds.forEach(id => console.log(id)); });
143. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); Observable
144. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); Observable
145. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); Observable
146. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); Observable
147. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); Observable
148. Cancellation
149. const promise = fetchOrders() .then((orders) => { orders.forEach((order) => { console.log(order); }); }); promise.cancel(); Promise
150. const promise = fetchOrders() .then((orders) => { orders.forEach((order) => { console.log(order); }); }); promise.cancel(); Promise
151. const promise = fetchOrders() .then((orders) => { orders.forEach((order) => { console.log(order); }); }); promise.cancel(); Promise ×
152. const subscription = fetchOrders() .subscribe((orders) => { orders.forEach((order) => { console.log(order); }); }); subscription.unsubscribe(); Observable
153. const subscription = fetchOrders() .subscribe((orders) => { orders.forEach((order) => { console.log(order); }); }); subscription.unsubscribe(); Observable Cancel request
154. Lazy Subscriptions
155. const p1 = fetchOrders(); const p2 = fetchOrders(); const p3 = fetchOrders(); Promises
156. const p1 = fetchOrders(); const p2 = fetchOrders(); const p3 = fetchOrders(); Promises Immediate
157. Observables const o1 = fetchOrders(); const o2 = fetchOrders(); const o3 = fetchOrders(); o1.subscribe(); o2.subscribe(); o3.subscribe();
158. Observables Lazy const o1 = fetchOrders(); const o2 = fetchOrders(); const o3 = fetchOrders(); o1.subscribe(); o2.subscribe(); o3.subscribe();
159. Observables Lazy const o1 = fetchOrders(); const o2 = fetchOrders(); const o3 = fetchOrders(); o1.subscribe(); o2.subscribe(); o3.subscribe(); Issue Request
160. const o1 = fetchOrders(); o1.subscribe(); o1.subscribe(); o1.subscribe();
161. const o1 = fetchOrders(); o1.subscribe(); o1.subscribe(); o1.subscribe(); Pure, shareable value
162. const o1 = fetchOrders(); o1.subscribe(); o1.subscribe(); o1.subscribe(); Issue new request with same observable Pure, shareable value
163. Create HTTP Requests
164. function fetchOrders() { return Observable.ajax.get('/orders'); } Built-in AJAX
165. function fetchOrders() { return Observable.ajax.get('/orders'); } Built-in AJAX
166. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation
167. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation
168. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation
169. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation
170. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation
171. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation
172. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Custom Observable Creation ?
173. Observable.of(1, 2, 3) .subscribe(x => console.log(x));
174. Observable.of(1, 2, 3) .subscribe(x => console.log(x)); Subscriber (or Observer)
175. Observable.of(1, 2, 3) .subscribe({ next: x => console.log(x), }); Subscriber Object
176. Observable.of(1, 2, 3) .subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), }); // 1 // 2 // 3 // Done!
177. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } fetchOrders().mergeAll().subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), });
178. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } fetchOrders().mergeAll().subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), });
179. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } fetchOrders().mergeAll().subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), });
180. Error Handling
181. const source = Observable.create((subscriber) => { subscriber.next(1); subscriber.error(new Error('Uh oh')); subscriber.next(2); }); source.subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), error: e => console.error(e), }); // 1 // Error: Uh oh
182. const source = Observable.create((subscriber) => { subscriber.next(1); subscriber.error(new Error('Uh oh')); subscriber.next(2); }); source.subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), error: e => console.error(e), }); // 1 // Error: Uh oh
183. const source = Observable.create((subscriber) => { subscriber.next(1); subscriber.error(new Error('Uh oh')); subscriber.next(2); }); source.subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), error: e => console.error(e), }); // 1 // Error: Uh oh
184. const source = Observable.create((subscriber) => { subscriber.next(1); subscriber.error(new Error('Uh oh')); subscriber.next(2); }); source.subscribe({ next: x => console.log(x), complete: () => console.log('Done!'), error: e => console.error(e), }); // 1 // Error: Uh oh Never called
185. const source = Observable.create((subscriber) => { subscriber.next(1); subscriber.error(new Error('Uh oh')); subscriber.next(2); }); source.subscribe(x => console.log(x)); No error handler, so what happens?
186. Error Error Error Error Promises
187. const source = Observable.create((subscriber) => { subscriber.next(1); subscriber.error(new Error('Uh oh')); subscriber.next(2); }); source.subscribe(x => console.log(x)); No swallowed errors!
188. fetchOrders() .catch((e) => { logError(e); return legacyFetchOrders() .catch((e2) => { logError(e2); return Observable.of([]); }) }) .subscribe(x => console.log(x)); Catching
189. fetchOrders() .catch((e) => { logError(e); return legacyFetchOrders() .catch((e2) => { logError(e2); return Observable.of([]); }) }) .subscribe(x => console.log(x)); Catching
190. fetchOrders() .catch((e) => { logError(e); return legacyFetchOrders() .catch((e2) => { logError(e2); return Observable.of([]); }) }) .subscribe(x => console.log(x)); Catching
191. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([])
192. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ✓
193. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ✓
194. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([])
195. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ×
196. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ×
197. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ×
198. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) × ✓
199. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) × ✓
200. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([])
201. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ×
202. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ×
203. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) ×
204. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) × ×
205. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) × ×
206. console.log catch subscribe legacyFetchOrders() fetchOrders() catch Observable.of([]) × × []
207. Hot Cold vs.
208. Observable creates the source Cold
209. Best for one-off unique requests. Cold
210. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); }
211. function fetchOrders() { return Observable.create((subscriber) => { fetchOrdersFromDb((orders) => { subscriber.next(orders); subscriber.complete(); }); }); } Resource requested/created at subscription time
212. WebSockets
213. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }); }
214. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }); }
215. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }); }
216. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }); }
217. const stream = ordersStream(); stream.subscribe(x => console.log(x));
218. subscriber .next(data) WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x));
219. subscriber .next(data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x));
220. subscriber .next(data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x));
221. subscriber .next(data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x));
222. subscriber .next(data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x));
224. .subscribe(...)
225. .subscribe(...) .subscribe(...)
226. .subscribe(...) .subscribe(...) .subscribe(...)
227. .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...)
228. .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...)
229. .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...)
230. .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...)
231. .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...) .subscribe(...)
232. Hot Observable closes over the source
233. Hot Best for multicasting and sharing resources.
234. const url = 'ws://example.com/orders'; const socket = new WebSocket(url); function ordersStream() { return Observable.create((subscriber) => { socket.addEventListener('message', (data) => { subscriber.next(data); }); }); }
235. const url = 'ws://example.com/orders'; const socket = new WebSocket(url); function ordersStream() { return Observable.create((subscriber) => { socket.addEventListener('message', (data) => { subscriber.next(data); }); }); } Resource created outside subscription
236. const url = 'ws://example.com/orders'; const socket = new WebSocket(url); function ordersStream() { return Observable.create((subscriber) => { socket.addEventListener('message', (data) => { subscriber.next(data); }); }); } Close over existing resource when subscribing
237. const stream = ordersStream(); const sub1 = stream.subscribe(x => console.log(x)); const sub2 = stream.subscribe(x => console.log(x));
238. Shared stream of data const stream = ordersStream(); const sub1 = stream.subscribe(x => console.log(x)); const sub2 = stream.subscribe(x => console.log(x));
239. subscriber .next(event.data) WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x))
240. subscriber .next(event.data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x))
241. subscriber .next(event.data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x))
242. subscriber .next(event.data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x))
243. subscriber .next(event.data) ... WebSocket Subscribed Observable console.log stream.subscribe(x => console.log(x))
244. subscriber .next(event.data) WebSocket Subscribed Observable console.log New Subscribed Observable console.log stream.subscribe(...) stream.subscribe(...)
245. subscriber .next(event.data) WebSocket Subscribed Observable console.log ... New Subscribed Observable console.log stream.subscribe(...) stream.subscribe(...)
246. subscriber .next(event.data) WebSocket Subscribed Observable console.log ... New Subscribed Observable console.log stream.subscribe(...) stream.subscribe(...)
247. subscriber .next(event.data) WebSocket Subscribed Observable console.log ... New Subscribed Observable console.log ... stream.subscribe(...) stream.subscribe(...)
248. subscriber .next(event.data) ... WebSocket Subscribed Observable console.log ... New Subscribed Observable console.log stream.subscribe(...) stream.subscribe(...)
249. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }) .share(); } Share It
250. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }) .share(); } Share It
251. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); }) .share(); } Share It
252. const stream = ordersStream(); const sub1 = stream.subscribe(x => console.log(x)); const sub2 = stream.subscribe(x => console.log(x));
253. Shared stream of data too const stream = ordersStream(); const sub1 = stream.subscribe(x => console.log(x)); const sub2 = stream.subscribe(x => console.log(x));
254. Clean Up
255. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); return () => { socket.close(); }; }) .share(); }
256. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); return () => { socket.close(); }; }) .share(); } Called when all subscriptions unsubscribe
257. function ordersStream() { return Observable.create((subscriber) => { const url = 'ws://example.com/orders'; const socket = new WebSocket(url); socket.addEventListener('message', (data) => { subscriber.next(data); }); return () => { socket.close(); }; }) .share(); } Close socket, deallocate resources, etc.
258. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); Recall
259. fetchOrders() .mergeAll() .filter( order => order.customerName === 'Tucker' ) .map(order => order.id) .subscribe(id => console.log(id)); ? Recall
260. Observable.of([1, 2, 3]) .subscribe(x => console.log(x)); // [ 1, 2, 3 ]
261. Observable.of([1, 2, 3]) .subscribe(x => console.log(x)); // [ 1, 2, 3 ]
262. Observable.of([1, 2, 3]) .subscribe(x => console.log(x)); // [ 1, 2, 3 ]
263. Observable.of([1, 2, 3]) .mergeAll() .subscribe(x => console.log(x)); // 1 // 2 // 3
264. Observable.of([1, 2, 3]) .mergeAll() .subscribe(x => console.log(x)); // 1 // 2 // 3 Flatten
265. Observable.of([1, 2, 3]) .mergeAll() .subscribe(x => console.log(x)); // 1 // 2 // 3
266. subscribe mergeAll console.log [1, 2, 3]
267. subscribe mergeAll console.log [1, 2, 3]
268. subscribe mergeAll console.log [2, 3] 1
269. subscribe mergeAll console.log [2, 3] 1
270. subscribe mergeAll console.log [3] 2
271. subscribe mergeAll console.log [3] 2
272. subscribe mergeAll console.log 3
273. subscribe mergeAll console.log 3
274. Higher Order Observables
275. Observable.of(1, 2, 3) .map(n => n * 2) .subscribe(x => console.log(x)); // 2 // 4 // 6
276. Observable.of(1, 2, 3) .map(n => n * 2) .subscribe(x => console.log(x)); // 2 // 4 // 6 Delay?
277. Observable.of(1, 2, 3) .map(n => Observable.of(n * 2)) .subscribe(x => console.log(x)); // ScalarObservable { value: 2 } // ScalarObservable { value: 4 } // ScalarObservable { value: 6 }
278. Observable.of(1, 2, 3) .map(n => Observable.of(n * 2)) .subscribe(x => console.log(x)); // ScalarObservable { value: 2 } // ScalarObservable { value: 4 } // ScalarObservable { value: 6 }
279. Observable.of(1, 2, 3) .map(n => Observable.of(n * 2)) .subscribe(x => console.log(x)); // ScalarObservable { value: 2 } // ScalarObservable { value: 4 } // ScalarObservable { value: 6 }
280. Observable.of(1, 2, 3) .map(n => Observable.of(n * 2)) .mergeAll() .subscribe(x => console.log(x)); // 2 // 4 // 6
281. Observable.of(1, 2, 3) .map(n => Observable.of(n * 2)) .mergeAll() .subscribe(x => console.log(x)); // 2 // 4 // 6
282. Observable.of(1, 2, 3) .mergeMap(n => Observable.of(n * 2)) .subscribe(x => console.log(x)); // 2 // 4 // 6
283. subscribe mergeMap console.log Observable.of(n * 2)
284. subscribe mergeMap console.log Observable.of(n * 2) 1
285. subscribe mergeMap console.log Observable.of(n * 2) 1
286. subscribe mergeMap console.log Observable.of(n * 2) Observable.of(2) subscribe 2
287. subscribe mergeMap console.log Observable.of(n * 2) Observable.of(2) subscribe 2
288. subscribe mergeMap console.log Observable.of(n * 2) Observable.of(2) subscribe 2
289. Observable.of(1) .delay(1000) .subscribe(x => console.log(x)) // <tick> // 1
290. Observable.of(1) .delay(1000) .subscribe(x => console.log(x)) // <tick> // 1
291. Observable.of(1) .delay(1000) .subscribe(x => console.log(x)) // <tick> // 1
292. Observable.of(1) .delay(1000) .subscribe(x => console.log(x)) // <tick> // 1
293. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) )) .subscribe(x => console.log(x)); // <tick> // 2 // 4 // 6
294. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) )) .subscribe(x => console.log(x)); // <tick> // 2 // 4 // 6
295. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) )) .subscribe(x => console.log(x)); // <tick> // 2 // 4 // 6
296. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) )) .subscribe(x => console.log(x)); // <tick> // 2 // 4 // 6 ?
297. Concurrency
298. subscribe mergeMap console.log Observable.of(n * 2)
299. subscribe mergeMap console.log Observable.of(n * 2) 1
300. subscribe mergeMap console.log Observable.of(n * 2) 1
301. subscribe mergeMap console.log Observable.of(n * 2) 2
302. subscribe mergeMap console.log Observable.of(n * 2) 2 2
303. subscribe mergeMap console.log Observable.of(n * 2) 2 2
304. subscribe mergeMap console.log Observable.of(n * 2) 2 4
305. subscribe mergeMap console.log Observable.of(n * 2) 3 2 4
306. subscribe mergeMap console.log Observable.of(n * 2) 3 2 4
307. subscribe mergeMap console.log Observable.of(n * 2) 2 4 6
308. subscribe mergeMap console.log Observable.of(n * 2) 2 4 6
309. subscribe mergeMap console.log Observable.of(n * 2) 4 6 2
310. subscribe mergeMap console.log Observable.of(n * 2) 6 4
311. subscribe mergeMap console.log Observable.of(n * 2) 6
312. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) ), 1) .subscribe(x => console.log(x)); // <tick> // 2 // <tick> // 4 // <tick> // 6
313. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) ), 1) .subscribe(x => console.log(x)); // <tick> // 2 // <tick> // 4 // <tick> // 6
314. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) ), 1) .subscribe(x => console.log(x)); // <tick> // 2 // <tick> // 4 // <tick> // 6
315. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) ), 1) .subscribe(x => console.log(x)); // <tick> // 2 // <tick> // 4 // <tick> // 6
316. Observable.of(1, 2, 3) .mergeMap(n => ( Observable.of(n * 2).delay(1000) ), 1) .subscribe(x => console.log(x)); // <tick> // 2 // <tick> // 4 // <tick> // 6
317. Observable.of(1, 2, 3) .concatMap(n => ( Observable.of(n * 2).delay(1000) )) .subscribe(x => console.log(x)); // <tick> // 2 // <tick> // 4 // <tick> // 6
318. subscribe concatMap console.log Observable.of(n * 2)
319. subscribe console.log Observable.of(n * 2) 1 concatMap
320. subscribe console.log Observable.of(n * 2) 1 concatMap
321. subscribe console.log Observable.of(n * 2) 2 concatMap
322. subscribe console.log Observable.of(n * 2) 2 concatMap
323. subscribe console.log Observable.of(n * 2) 2 concatMap
324. subscribe console.log Observable.of(n * 2) 2 concatMap
325. subscribe console.log Observable.of(n * 2) 2 concatMap
326. subscribe console.log Observable.of(n * 2) 4 concatMap
327. subscribe console.log Observable.of(n * 2) 4 concatMap
328. subscribe console.log Observable.of(n * 2) 4 concatMap
329. subscribe console.log Observable.of(n * 2) 3 concatMap
330. subscribe console.log Observable.of(n * 2) 3 concatMap
331. subscribe console.log Observable.of(n * 2) 6 concatMap
332. subscribe console.log Observable.of(n * 2) 6 concatMap
333. subscribe console.log Observable.of(n * 2) 6 concatMap
334. Rate Limiting
335. Observable.of(1, 2, 3) .concatMap((id) => { const url = `/orders/${id}`; return Observable.ajax.get(url) .delay(1000); }) .pluck('response') .bufferCount(3) .subscribe(x => console.log(x)); // [ { id: '1', name: 'Order 1' }, // { id: '2', name: 'Order 2' }, // { id: '3', name: 'Order 3' } ]
336. Observable.of(1, 2, 3) .concatMap((id) => { const url = `/orders/${id}`; return Observable.ajax.get(url) .delay(1000); }) .pluck('response') .bufferCount(3) .subscribe(x => console.log(x)); // [ { id: '1', name: 'Order 1' }, // { id: '2', name: 'Order 2' }, // { id: '3', name: 'Order 3' } ]
337. Observable.of(1, 2, 3) .concatMap((id) => { const url = `/orders/${id}`; return Observable.ajax.get(url) .delay(1000); }) .pluck('response') .bufferCount(3) .subscribe(x => console.log(x)); // [ { id: '1', name: 'Order 1' }, // { id: '2', name: 'Order 2' }, // { id: '3', name: 'Order 3' } ]
338. Observable.of(1, 2, 3) .concatMap((id) => { const url = `/orders/${id}`; return Observable.ajax.get(url) .delay(1000); }) .pluck('response') .bufferCount(3) .subscribe(x => console.log(x)); // [ { id: '1', name: 'Order 1' }, // { id: '2', name: 'Order 2' }, // { id: '3', name: 'Order 3' } ]
339. Observable .ajax .get(url) subscribe console.log pluck bufferCount 'response' 3 concatMap
340. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 3 1
341. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 3
342. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 3 R Rate limit next request
343. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 3 R
344. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 3 R
345. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 3 O1
346. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 2 O1
347. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 2 O1 +
348. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 1 O1 O2
349. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 1 O1 O2 +
350. Observable .ajax .get(url) subscribe concatMap console.log pluck bufferCount 'response' 0 O1 O2 O3
351. Observable .ajax .get(url) subscribe concatMap console.log […] pluck bufferCount 'response' 0
352. Sequential vs. Parallel
353. const promise = Promise.all([ fetchOrder(1), fetchOrder(2), fetchOrder(3), ]); promise.then(x => console.log(x)); // [ { id: '1', name: 'Order 1' }, // { id: '2', name: 'Order 2' }, // { id: '3', name: 'Order 3' } ]
354. const promise = Promise.all([ fetchOrder(1), fetchOrder(2), fetchOrder(3), ]); promise.then(x => console.log(x)); // [ { id: '1', name: 'Order 1' }, // { id: '2', name: 'Order 2' }, // { id: '3', name: 'Order 3' } ]
355. const source = Observable.forkJoin( fetchOrder(1), fetchOrder(2), fetchOrder(3) ); source.subscribe(x => console.log(x)); // [ { response: { id: 1, name: 'Order 1' } }, // { response: { id: 2, name: 'Order 2' } }, // { response: { id: 3, name: 'Order 3' } } ]
356. const source = Observable.forkJoin( fetchOrder(1), fetchOrder(2), fetchOrder(3) ); source.subscribe(x => console.log(x)); // [ { response: { id: 1, name: 'Order 1' } }, // { response: { id: 2, name: 'Order 2' } }, // { response: { id: 3, name: 'Order 3' } } ]
357. const source = Observable.forkJoin( fetchOrder(1), fetchOrder(2), fetchOrder(3) ); source.subscribe(x => console.log(x)); // [ { response: { id: 1, name: 'Order 1' } }, // { response: { id: 2, name: 'Order 2' } }, // { response: { id: 3, name: 'Order 3' } } ] Observable or Promise
358. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin subscribe console.log
359. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin subscribe console.log R1
360. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin R1 subscribe console.log
361. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin R1 subscribe console.log R3
362. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin R1 R3 subscribe console.log
363. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin R1 R3 subscribe console.log R2
364. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin R1 R2 R3 subscribe console.log
365. fetchOrder(1) fetchOrder(2) fetchOrder(3) forkJoin subscribe console.log R1 R2 R3
366. Tip of the iceberg
367. • Declarative, lazy operations • Expressive event management • No more error swallowing • Rate limiting and concurrent processing
368. • Declarative, lazy operations • Expressive event management • No more error swallowing • Rate limiting and concurrent processing ✓ ✓ ✓ ✓
369. github.com/ReactiveX/rxjs reactivex.io/rxjs RxJS
370. Observables ECMAScript Proposal:
 github.com/tc39/proposal-observable Another spec implementation:
 github.com/zenparsing/zen-observable
371. Thanks! Jeremy Fairbank @elpapapollo / jfairbank Slides: bit.ly/rxjs-connect