RxSwift: Deep Cuts

1. RxSwift: 
 Deep Cuts Krzysztof Siejkowski
2. Alex is an iOS developer
3. Reactive programming
4. RxSwift ❤
6. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
7. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
8. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
9. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
10. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
11. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
12. #1 What references 
 does it carry?
13. Observable .just(42) Just
14. Observable .just(42) .filter { $0 > 30 } Just Filter
15. Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } Just Filter Map
16. Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } Just Filter Map Distinct UntilChanged
17. let observable: Observable<String> = Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } Just Filter Map observable observable
18. environment Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } Just Filter Map Distinct UntilChanged filter closure compare closure map closure
19. Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) } Just Filter Map Distinct UntilChanged filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink
20. Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) } Just Filter Map Distinct UntilChanged filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink
21. Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) } Just Filter Map Distinct UntilChanged filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Observer
22. filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Just Sink Disposer Filter Sink Disposer Map Sink Disposer DUC Sink Disposer Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) }
23. reference cycles filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Just Sink Disposer Filter Sink Disposer Map Sink Disposer DUC Sink Disposer
24. filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Just Sink Disposer Filter Sink Disposer Map Sink Disposer disposable let disposable: Disposable = Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) }
25. disposable .dispose() filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Just Sink Disposer Filter Sink Disposer Map Sink Disposer disposable
26. .dispose() — called externally 
 on subscription disposable — called internally 
 when event stream ends
27. filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Just Sink Disposer Filter Sink Disposer Map Sink Disposer disposable observe closure let disposable = Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) }
28. environment let disposable = Observable .just(42) .filter { $0 > 30 } .map { “\($0)” } .distinctUntilChanged { $0 !== $1 } .subscribe { print($0) } filter closure compare closure map closure Just Sink Filter Sink Map Sink DUC Sink Observer Just Sink Disposer Filter Sink Disposer Map Sink Disposer disposable observe closure
29. Memory leaks and extended lifetime
30. Dispose your subscriptions
31. Watch your references
32. observable .map { [weak self] in self!?.transform($0) } .subscribe { [weak self] in self!?.react(to: $0) } observable .map { [unowned self] in self.transform($0) } .subscribe { [unowned self] in self.react(to: $0) } .disposed( by: self.disposeBag )
33. !// instance method func foo(_ bar: Int) !-> Int !// instance scope Observable.just(42) .map(foo) !// instance method func foo(_ bar: Int) !-> Int !// instance scope Observable.just(42) .map { self.foo($0) }
34. dataProvider.refreshData() .subscribe( onNext: { [weak self] in self!?.update(with: $0) }, onError: { [weak self] in if let error = $0 as? ReasonableError { self!?.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
35. dataProvider.refreshData() .subscribe( onNext: { [unowned self] in self.update(with: $0) }, onError: { [unowned self] in if let error = $0 as? ReasonableError { self.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
36. dataProvider.refreshData() .subscribe( onNext: { [unowned self] in self.update(with: $0) }, onError: { [unowned self] in if let error = $0 as? ReasonableError { self.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
37. #2 What thread does it do the work on?
38. Schedulers define context of work execution
39. ImmediateSchedulerType func schedule<StateType>( _ state: StateType, action: @escaping (StateType) !-> Disposable ) !-> Disposable

40. MainScheduler let mainQueue = DispatchQueue.main !// simplified essence of schedule method if DispatchQueue.isMain { action(state) } else { mainQueue.async { action(state) } }
41. SerialDispatchQueueScheduler !// internal serial dispatch queue of given properties let queue = DispatchQueue.global(qos: qos.qosClass) !// simplified essence of schedule method queue.async { action(state) }
42. OperationQueueScheduler !// operation queue provided by client in the initializer let operationQueue: OperationQueue !// simplified essence of schedule method operationQueue.addOperation( BlockOperation { action(state)) } )
43. Also in schedulers • QOS • serial / concurrent • immediate / delayed • driven by client
44. Scheduler-using operators interval(1, scheduler) delay(2, scheduler) throttle(3, scheduler)
45. Scheduler-agnostic operators map { foo($0) } flatMap { bar($0) } filter { $0 !== wanted }
46. Scheduler-defining operators observeOn(scheduler) subscribeOn(scheduler)
47. Observable .just(42) .filter { $0 > 33 } .map { "\($0)" } .subscribe { print($0) } .disposed(by: disposeBag) Subscription thread
48. Observable .just(42) .observeOn(greenScheduler) .filter { $0 > 33 } .map { "\($0)" } .subscribe { print($0) } .disposed(by: disposeBag) Subscription thread Green scheduler
49. Observable .just(42) .observeOn(greenScheduler) .filter { $0 > 33 } .observeOn(redScheduler) .map { "\($0)" } .observeOn(greenScheduler) .subscribe { print($0) } .disposed(by: disposeBag) Subscription thread Red scheduler Green scheduler Green scheduler
50. Observable .just(42) .observeOn(greenScheduler) .filter { $0 > 33 } .observeOn(redScheduler) .subscribeOn(blueScheduler) .map { "\($0)" } .observeOn(greenScheduler) .subscribe { print($0) } .disposed(by: disposeBag) Blue scheduler Red scheduler Green scheduler Green scheduler
51. Observable .just(42) .observeOn(greenScheduler) .filter { $0 > 33 } .subscribeOn(blueScheduler) .observeOn(redScheduler) .subscribeOn(redScheduler) .map { "\($0)" } .observeOn(greenScheduler) .subscribeOn(greenScheduler) .subscribe { print($0) } .disposed(by: disposeBag) Blue scheduler Red scheduler Green scheduler Green scheduler
52. Observable .just( 42, scheduler: blueScheduler ) .observeOn(greenScheduler) .filter { $0 > 33 } .observeOn(redScheduler) .subscribeOn(redScheduler) .map { "\($0)" } .observeOn(greenScheduler) .subscribe { print($0) } .disposed(by: disposeBag) Blue scheduler Red scheduler Green scheduler Green scheduler
53. Schedulers used for: • generation • internal work
54. Schedule-using for generation !// Scheduler to send elements on. .fromScheduled([1, 2, 3], greenScheduler) !// Scheduler to run the producer loop on. .repeatElement(1, blueScheduler) !// Scheduler to run the timer on. .interval(1, redScheduler)
55. Schedule-using for internal work !// Scheduler to run the throttle timers on. .throttle(1, blueScheduler) !// Scheduler to run the subscription delay timer on. .delay(1, greenScheduler) !// Scheduler to run buffering timers on. .buffer(timeSpan: 1, count: 3, scheduler: redScheduler)
56. Delivery thread may change between events
57. Observable<Int> .merge([ Observable<Int>.interval(3, redScheduler), Observable<Int>.interval(5, greenScheduler), ]) .subscribe { [unowned self] in !// sometimes red, sometimes green self.work(with: $0) } .disposed(by: disposeBag)
58. Read the docs 
 and 
 the source code
59. When in doubt, ensure with observeOn
60. RxCocoa traits guarantee MainScheduler
61. dataProvider.refreshData() .observeOn(MainScheduler.instance) .subscribe( onNext: { [unowned self] in self.update(with: $0) }, onError: { [unowned self] in if let error = $0 as? ReasonableError { self.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
62. dataProvider.refreshData() .observeOn(MainScheduler.instance) .subscribe( onNext: { [unowned self] in self.update(with: $0) }, onError: { [unowned self] in if let error = $0 as? ReasonableError { self.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
63. #3 What protocol does it follow?
64. .next(data) .completed .error(error)
65. Different protocols may need different subscriptions
66. Does it finish?
67. Finishing !// you may ignore disposable (though I recommend not to!) _ = observable .subscribe(onNext: { [weak self] in self!?.work(on: $0) })
68. Not finishing let disposable = observable .subscribe(onNext: { [weak self] in self!?.work(on: $0) }) !// ensure subscription disposing disposable.dispose() disposable.disposed(by: disposeBag)
69. How long 
 does it last?
70. One-off let serial = SerialDisposable() !// call each time you want the fresh data func fetchFreshData() { serial.disposable = observable .subscribe(onNext: { [unowned self] in self.work(on: $0) }) }
71. Updating let disposeBag = DisposeBag() !// call only once in the object lifetime func listenForFreshData() { observable .subscribe(onNext: { [unowned self] in self.work(on: $0) }) .disposed(by: disposeBag) }
72. When and how does it 
 start emitting?
73. Immediately & synchronously !// if not ready to handle data at the `subscribe` time dataProvider .data() .skip(1) .subscribe(onNext: { [unowned self] in self.work(on: $0) }) .disposed(by: disposeBag)
74. After some time & asynchronously !// provide separate method for fetching initial data let initialData = dataProvider.getInitialData() dataProvider .dataUpdates() .subscribe(onNext: { [unowned self] in self.work(on: $0) }) .disposed(by: disposeBag)
75. Does it cache 
 the events?
76. Not caching let observable = dataProvider .fetchHugeAmountOfDataFromNetwork() !// costly network request with fresh data observable.subscribe() !// costly network request with fresh data observable.subscribe()
77. Caching let observable = dataProvider .fetchHugeAmountOfDataFromNetwork() .shareReplay(1) !// costly network request with fresh data observable.subscribe() !// no network refresh but also old data observable.subscribe()
78. Also in protocols • error types • time relations • many more
79. Expose the protocol via the type system
80. RxCocoa traits • Driver • Signal • ControlProperty
81. RxSwift traits • Single • Completable • Maybe
82. Document the protocol
83. !/* Returns an Observable that emits at most three times, starting with the first event emitted immediately and synchronously upon subscription. Times of other two events are not guaranteed. May not complete, but never errors out. Doesn’t cache any data. !*/ public func thirdTimeLucky() !-> Observable<Data>
84. Conventionalize the protocol
85. final class DataProvider { private let proxySubject = PublishSubject<Data>() var data: Observable<Data> { return proxySubject.asObservable() } func refreshData() !-> Observable<Void> { return networkService .requestData() .do(onNext: { proxySubject.onNext($0) } .map { _ in } }
86. Limit the scope of the protocol
87. Code distance
88. Network Persistence Domain services View Models View long code distance
89. Network Persistence Domain services View Models View short code distance
90. Short code distance !== easier reasoning
91. Alex has found 
 the bug
92. dataProvider.refreshData() .observeOn(MainScheduler.instance) .subscribe( onNext: { [unowned self] in self.update(with: $0) }, onError: { [unowned self] in if let error = $0 as? ReasonableError { self.showUserMessage(with: error.reason) } } ) .disposed(by: disposeBag)
94. RxSwift ❤
95. Thank you 
 and happy coding! siejkowski.net