Rx by example – Volume 3 (the multicast edition)

Given at 360|AnDev

Video: https://academy.realm.io/posts/360-andev-2017-kaushik-gopal-rxjava-by-example-multicasting/

As developer’s understanding of RxJava has matured, they’ve started to peel back the layers and unleash its true power and potential.

One such power is “Multicasting” where you get to share work across your app and reuse a whole bunch of stuff. This is really hard to do in general but Rx makes it really really easy.

Rx itself though has a steep learning curve. But the easiest way to grasp the concepts and features of Rx, is through examples. I’ve given talks in the past explaining Rx usage purely through examples and folks have found this to be the easiest way to warm up to Rx.

Transcript

1. Rx by example
 The Multicast edition Vol. 3
2. Multicasting What is Why How
3. Multicasting 2 ways to do ConnectableObservables Subjects
4. Multicasting with ConnectableObservables
5. val source: Observable<Long> = Observable.interval(1, TimeUnit.SECONDS) source.subscribe { Log.v(TAG, "subscriber 1 : received event $it") } subscriber 1 : received event 0 subscriber 2 : received event 0 subscriber 1 : received event 1 subscriber 2 : received event 1 subscriber 1 : received event 2 subscriber 2 : received event 2 subscriber 1 : received event 3 subscriber 2 : received event 3 subscriber 1 : received event 4 subscriber 2 : received event 4 source.subscribe { Log.v(TAG, "subscriber 2 : received event $it") }
6. val source: Observable<Long> = Observable.interval(1, TimeUnit.SECONDS) Thread.sleep(1000) source.subscribe { Log.v(TAG, "subscriber 1 : received event $it") } source.subscribe { Log.v(TAG, "subscriber 2 : received event $it") } subscriber 1 : received event 0 Skipped 59 frames! app may be doing too much work on its main thread. subscriber 1 : received event 1 subscriber 2 : received event 0 subscriber 1 : received event 2 subscriber 2 : received event 1 subscriber 1 : received event 3 subscriber 2 : received event 2 subscriber 1 : received event 4 subscriber 2 : received event 3
7. val source: Observable<Long> = Observable.interval(1, TimeUnit.SECONDS) Thread.sleep(1000) source.subscribe { Log.v(TAG, "subscriber 1 : received event $it") } source.subscribe { Log.v(TAG, "subscriber 2 : received event $it") } ConnectableObservable<Long> = .publish() source.connect() Skipped 59 frames! app may be doing too much work on its main thread. subscriber 1 : received event 0 subscriber 2 : received event 0 subscriber 1 : received event 1 subscriber 2 : received event 1 subscriber 1 : received event 2 subscriber 2 : received event 2 subscriber 1 : received event 3 subscriber 2 : received event 3
8. .publish().refcou .replay().autoCo .share().cache( .replayingShar blish().refcoun eplay().autoConnect(n share().cache() .replayingShare() h().refcount() y().autoConnect(n) ).cache() are() .publish().refcount .replay().autoConn re().cache() ngShare( Multicasting operator permutations
9. val sourceObservable = Observable.interval(1, TimeUnit.SECONDS)
 .doOnSubscribe{ log("observable (subscribed)") }
 .doOnDispose{ log("observable (disposed)") }
 disposable1?.let {
 it.dispose()
 log("subscriber 1 disposed")
 disposable1 = null
 return
 } @OnClick(R.id.btn_1)
 fun onBtn1Click() { disposable1 = sharedObservable.subscribe({ long -> _log(“subscriber 1: onNext $long") }) } https://github.com/kaushikgopal/RxJava-Android-Samples/tree/kg/feat/connect_playground
10. val sourceObservable = Observable.interval(1, TimeUnit.SECONDS) sourceObservable.connect() .doOnSubscribe{ log("observable (subscribed)") }
 .doOnDispose{ log("observable (disposed)") } .publish() https://github.com/kaushikgopal/RxJava-Android-Samples/tree/kg/feat/connect_playground
11. val sourceObservable = Observable.interval(1, TimeUnit.SECONDS) .doOnSubscribe{ log("observable (subscribed)") }
 .doOnDispose{ log("observable (disposed)") } sourceObservable .refCount() .publish() https://github.com/kaushikgopal/RxJava-Android-Samples/tree/kg/feat/connect_playground
12. sourceObservable .publish().refCount()
13. sourceObservable .publish().refCount() subscribe 1 subscriber so connect automatically 0 0 onNext onNext 1 1 onNext onNext subscribe 2 subscribers 2 onNext onNext 2 onNext 2 3 onNext onNext 3 onNex 3 refcount()
14. ribers 2 onNext onNext 2 onNext 2 3 onNext onNext 3 onNext 3 refcount() unsubscribe X unsubscribe 1 subscriber left 4 4 onNext onNext unsubscribe X X 0 subscribers left
15. ext refcount() unsubscribe X unsubscribe 1 subscriber left 4 4 onNext onNext X X 0 subscribers left subscribe 0 0 onNext onNext YO LISTEN UP !! THIS IS IMPORTANT
16. sourceObservable .publish().autoConnect(2)
17. sourceObservable .publish().autoConnect(2) .autoConnect(2) subscribe 1 subscriber (waiting for 2) subscribe 2 subscribers 0 onNext onNext 0 onNext 0 1 onNext onNext 1 onNext 1 unsubscribe X unsubscribe 1 subscriber left
18. sourceObservable .publish().autoConnect(2) ext Next Next 1 onNext onNext 1 onNext 1 X unsubscribe 1 subscriber left .autoConnect(2) 2 2 onNext onNext unsubscribe X 0 subscribers left 3 4 YO Me AGAIN !! THIS IS IMPORTANT
19. sourceObservable .publish().autoConnect(2) be ber left 2 2 onNext onNext unsubscribe X 0 subscribers left 3 4 .autoConnect(2) subscribe 5 5 onNext onNext first event seen is 5
20. Recap .publish() .autoConnect(n) .refcount() Creates ConnectableObservable Requires .connect() or something else to kickstart Source subscription started right away Source is “restarted” when all subscribers leave and 
 new subscribers come in Source subscription started
 only after n subscribers start Source never stops once started (even if all subscribers leave)
21. .publish() Creates ConnectableObservable .replay(n) Creates ConnectableObservable Another way to create ConnectableObservables Start with last “n” events “n” defaults to all events
22. Pop Quiz ! .publish().refCount() .publish().autoConnect(1) Vs a.k.a .share()
23. Pop Quiz ! .replay(1).refCount() .replay(1).autoConnect(1) Vs ~ a.k.a .cache() *
24. sourceObservable .replay(1).refCount() .replay(1) .refCount() subscribe 1 subscriber so connect automatically 0 0 onNext onNext subscribe 2 subscribers onNext 0 unsubscribe X unsubscribe X X subscribe No replay!
25. sourceObservable .replay(1).refCount() .replay(1) .autoConnect() subscribe 1 subscriber so connect automatically 0 0 onNext onNext subscribe 2 subscribers onNext 0 unsubscribe X unsubscribe X subscribe onNext 0
26. Pop Quiz ! .replayingShare() = .replay(1).refcount() But remembers the last item even when restarted .replay(1).refCount()
27. Use cases Let’s get real
28. Location provider Use case 1 return RxLocation(context) .updates(locationRequest) // multicast? .publish() .autoConnect(n) .replay(1) .refcount() .replayingShare() .replay(1) .refcount()
29. Use case 2 Persist a network call beyond activity rotation ViewModels baby ...
30. Persist a network call beyond activity rotation class RotationPersistActivity : MyBaseActivity() { lateinit var sharedViewModel: SharedViewModel override fun onCreate(savedInstanceState: Bundle?) {
 super.onCreate(savedInstanceState)
 sharedViewModel = ViewModelProviders.of(activity) .get(SharedViewModel::class.java) 
 }
 
 @OnClick(R.id.btn_rotate_persist)
 fun displayEventsOnBtnClick() {
 // … compositeDisposables += sharedViewModel
 .networkRequest()
 .subscribe({ l ->
 Use case 2
31. class RotationPersistActivity : MyBaseActivity() { lateinit var sharedViewModel: SharedViewModel override fun onCreate(savedInstanceState: Bundle?) {
 super.onCreate(savedInstanceState)
 sharedViewModel = ViewModelProviders.of(activity) .get(SharedViewModel::class.java) 
 }
 
 @OnClick(R.id.btn_rotate_persist)
 fun displayEventsOnBtnClick() {
 // … compositeDisposables += sharedViewModel
 .networkRequest()
 .subscribe({ l ->
 log("Received element $l")
 }) } } Persist a network call beyond activity rotation Use case 2
32. Persist a network call beyond activity rotation override fun onCleared() {
 super.onCleared()
 // cleanup } Use case 2 
 fun networkRequest(): Observable<Data> {
 
 // We’ll fill this later } 
 class SharedViewModel : ViewModel() {

33. Use case 2 Persist a network call beyond activity rotation 
 fun networkRequest(): Observable<Data> {
 
 // We’ll fill this later } override fun onCleared() {
 super.onCleared()
 // cleanup } var sharedStream: Observable<Data> = someApi.makeRequest() //.map(…) .replay(1) .autoConnect() 
 class SharedViewModel : ViewModel() {
 return sharedStream
34. Persist a network call beyond activity rotation Use case 2 
 fun networkRequest(): Observable<Data> {
 
 } override fun onCleared() {
 super.onCleared()
 // cleanup } var sharedStream: Observable<Data> = someApi.makeRequest() //.map(…) .replay(1) .autoConnect() 
 class SharedViewModel : ViewModel() {
 return sharedStream
35. Persist a network call beyond activity rotation .replay(1) .autoConnect(1, disposable -> { myDisposable = disposable }) var myDisposable: Disposable? = null myDisposable?.dispose 
 class SharedViewModel : ViewModel() {
 var sharedStream: Observable<Data> = someApi.makeRequest() //.map(…) override fun onCleared() {
 super.onCleared()
 // cleanup } fun networkRequest(): Observable<Data> {
 return sharedStream } Use case 2
36. _paginator .map(items ->{ //do something }) .onBackPressureDrop() .doOnNext(
 i -> {
 _progressBar.setVisibility(View.VISIBLE);
 }) .concatMap(this::itemsFromNetworkCall) .observeOn(AndroidSchedulers.mainThread()) .map(items -> {
 _adapter.addItems(items);
 _adapter.notifyDataSetChanged();
 return items;
 })
 .subscribe(strings -> {
 _progressBar.setVisibility(View.INVISIBLE);
 });
 Use case 3 Reduce side effects
37. .concatMap(this::itemsFromNetworkCall) .observeOn(AndroidSchedulers.mainThread()) .map(items -> {
 _adapter.addItems(items);
 _adapter.notifyDataSetChanged();
 return items;
 })
 .subscribe(strings -> {
 _progressBar.setVisibility(View.INVISIBLE);
 });
 Use case 3 Reduce side effects .doOnNext(
 i -> {
 _progressBar.setVisibility(View.VISIBLE);
 }) _paginator .map(items ->{ //do something }) .onBackPressureDrop()
38. .doOnNext(
 i -> {
 _progressBar.setVisibility(View.VISIB }) Use case 3 Reduce side effects _paginator .map(items ->{ //do something }) .onBackPressureDrop() .publish() ConnectableObservable<Int> shared = shared shared .subscribe(
 i -> {
 _progressBar.setVisibility(View.VISIB }) .concatMap(this::itemsFromNetworkCall) .observeOn(AndroidSchedulers.mainThread()) .map(items -> {
 _adapter.addItems(items);
 _adapter.notifyDataSetChanged();
 return items;
 })
 .subscribe(strings -> {
 _progressBar.setVisibility(View.INVISIBLE);
 });

39. .concatMap(this::itemsFromNetworkCall) .observeOn(AndroidSchedulers.mainThread()) .map(items -> {
 _adapter.addItems(items);
 _adapter.notifyDataSetChanged();
 return items;
 })
 .subscribe(strings -> {
 _progressBar.setVisibility(View.INVISIBLE);
 });
 Use case 3 Reduce side effects _paginator .map(items ->{ //do something }) .onBackPressureDrop() .publish() ConnectableObservable<Int> shared = shared.connect(); .subscribe(
 i -> {
 _progressBar.setVisibility(View.VISIBLE);
 }) shared shared
40. ✌ fragmentedpodcast.com @kaushikgopal kaush.co We're hiring Android devs! tech.instacart.com
41. Guided Q&A
42. I'm worried about Multicasting leaking things around? Should I be? If multi-casted stream "errors"/"terminates" What happens to subscribers? How do I dispose of ConnectableObs. when using autoConnect? Where Multicasting begins (at which point of the Observable) before/after publish ? When should I use ConnectableObsv. vs Subjects ? 1 2 3 4 5