RxJava: A Stream of Joy and Woe

RxJava has become an invaluable tool to many Android developers, aiding in the composition of the several asynchronous systems we must deal with. However, with so much power RxJava can sometimes be the hammer we hold with almost everything looking like a nail. This talk aims to run through anecdotal examples of where RxJava has worked well, and where it maybe wasn’t the best idea.

Transcript

1. RxJava A Stream of Joy and Woe @chris_h_codes
2. Why have we chased the dragon?
3. <OUR CODE>
4. <OUR CODE>
5. <OUR CODE>
6. <OUR CODE>
7. <OUR CODE>
8. webServices.getSomething() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
9. webServices.getSomething() .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
10. webServices.getSomething() .flatMap { firstResult -> webServices.getSomethingElse(firstResult) } .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
11. webServices.getSomething() .flatMap { firstResult -> webServices.getSomethingElse(firstResult) } .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe ( { something -> // ... }, { error -> // ... } )
12. webServices.getSomething() .flatMap { firstResult -> webServices.getSomethingElse(firstResult) } .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe ( { something -> // ... }, { error -> // ... } ) OF TERMINAL EVENTS BEWARE
13. val getSomethingFromWeb = webServices.getSomething() .flatMap { firstResult -> webServices.getSomethingElse(firstResult) } .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread())
14. systemEventEmitter() .switchMap { getSomethingFromWeb.toObservable() } .withLatestFrom(somethingElse) { // ... }
15. interface WebServices { @GET("something") fun getSomething(): Single<Something>> }/
16. interface WebServices { @GET("something") fun getSomething(): Single<Result<Something>> }/
17. webServices.getSomething() .flatMap { firstResult -> webServices.getSomethingElse(firstResult) } .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
18. webServices.getSomething() .flatMap { firstResult -> webServices.getSomethingElse(firstResult) } .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
19. webServices.getSomething() .flatMap { firstResult -> if (!firstResult.isError()) { webServices.getSomethingElse(firstResult.response().body()) } else { Single.just(ERROR) }/ }/ .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
20. webServices.getSomething() .flatMap { firstResult -> if (!firstResult.isError() && firstResult.response().isSuccessful) { webServices.getSomethingElse(firstResult.response().body()) } else { Single.just(ERROR) }/ }/ .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
21. webServices.getSomething() .flatMap { firstResult -> if (!firstResult.isError() && firstResult.response().isSuccessful) { webServices.getSomethingElse(firstResult.response().body()) } else { Single.just(ERROR) }/ }/ .map { result -> transform(result) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { something -> // ... }/
22. suspend fun doSomeWork() { try { val firstResponse = webServices.getSomething().await() if (firstResponse.isSuccessful) { val something = firstResponse.body() val secondResponse = webServices.getSomethingElse(something).await() // .. } } catch (e: IOException) { // .. } }
23. <OUR CODE>
25. • Interfacing with an engine monitoring system • Over bluetooth • Which is emulating a serial connection • Enabling a (very slow) satellite internet service • Via SOAP • Sending location and engine data over that connection • Meanwhile monitoring the accelerometer to check for accidents
27. Monitoring for accidents
28. Z X
29. Y
30. android.hardware.Sensor.TYPE_GRAVITY XYZ XYZ XYZ XZ XZ XZ XYZ Check against some threshold ! Throttle detections
31. sensors.observeSensor(TYPE_GRAVITY) .distinctUntilChanged() .map { it.sensorEvent.values } .filter { it[0] > ROLLOVER_THRESHOLD || it[2] > ROLLOVER_THRESHOLD } .map { Emergency.ROLLOVER } .throttleFirst(THROTTLE_SECONDS, TimeUnit.SECONDS) .observeOn(RealRxSchedulers.mainThread()) .subscribe { }
32. • Location updates • Bluetooth payloads from engine • Managing output sockets as network conditions change • Shift status timers
33. RxPermissions .request(Manifest.permission.CAMERA) .subscribe { granted -> if (granted) { // I can control the camera now. } else { // Oh no, permission denied. } }
34. RxActivityResult.on(this) .startIntent(takePhoto) .subscribe { result, resultCode -> if (resultCode == RESULT_OK) { result.targetUi().showImage(data) } else { result.targetUi().printUserCancelled() } }
35. <OUR CODE>
36. 13:37 chris_h_codes
37. 13:37 chris_h_codes
38. 13:37 chris_h_codes
39. fun onViewAttached() { val user = userStore.getCurrentValue() avatarView.setImage(user.profilePic) usernameView.text = user.username }/
40. fun onViewAttached() { val user = userStore.getCurrentValue()() avatarView.setImage(user.profilePic) usernameView.text = user.username userStore.observe().subscribe { user -> avatarView.setImage(user.profilePic) usernameView.text = user.usernmame } }/
41. https:/ /github.com/Gridstone/RxStore
42. https:/ /github.com/Gridstone/RxStore val userStore = storeProvider.valueStore<User>(file, converter) userStore.put(user) val observeUser: Observable<User> = userStore.observe() val putUser: Single<User> = userStore.observePut(user)
43. webServices.getUser() .concatMap { user -> userStore.observePut(user) } .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread())
44. <OUR CODE>
45. Controller View
46. Controller View
47. State Controller View
48. Controller View State Event
49. Controller View Observable<Event> Observable<State>
50. val events: Observable<Event> = Observable.merge( loginButton.clicks().map { Event.AttemptLogin(username, password) }, cancelButton.clicks().map { Event.Cancel } ) https:/ /github.com/JakeWharton/RxBinding
51. Observables .combineLatest( usernameView.textChanges().map { it.isNotEmpty() }, passwordView.textChanges().map { it.isNotEmpty() }) { hasUsername, hasPassword -> hasUsername && hasPassword } .subscribe(loginButton::setEnabled) Username Password Login
52. Controller View Observable<Event> Observable<State>
55. View() onFinishInflate() onRestoreInstanceState() draw() display(state)
56. stateStream .subscribeOn(Schedulers.computation()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { state -> view.display(state) }/
57. stateStream .replay(1) .autoConnect() .subscribeOn(Schedulers.computation()) .observeOn(AndroidSchedulers.mainThread()) .subscribe { state -> view.display(state) }/
58. stateStream .subscribeOn(Schedulers.computation()) .observeOn(AndroidSchedulers.mainThread()) .replay(1) .autoConnect() .subscribe { state -> view.display(state) }/
59. stateStream
60. action state reduce(action, previousState)
61. Takeaways
62. Takeaways • If you only use Rx for callbacks with nice threading, consider coroutines • Beware terminal events • Be mindful of thread boundaries and state delivery • Don’t try to Rx-AllTheThings
64. RxJava A Stream of Joy and Woe chris_h_codes chris-horner chrishorner.codes