Advanced RxJava+Networking - DroidconNYC

Slides from my talk "Advanced Networking with RxJava + Retrofit" given at Droidcon NYC

Transcript

1. Networking with RxJava + Retrofit Beyond the canonical REST example Stephen D’Amico - @sddamico - Droidcon NYC - Sept 2017
2. The Basics I bet at some point in your journey of learning networking with RxJava + Retrofit, you’ve seen a code example that looks something like this... interface SeatGeekApi { @GET("events") Observable<EventsResponse> getUpcomingEvents(); } Retrofit retrofit = new Retrofit.Builder() .baseUrl("https://api.seatgeek.com/") .build(); SeatGeekApi api = retrofit.create(SeatGeekApi.class); api.getUpcomingEvents() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(events -> { // do something with events }, error -> { // handle errors });
3. But, what doesn't this handle? • Retrying requests when they fail • Dealing with specific response codes • Showing loading indicators in your UI • Configuration changes/subscription management • Probably other cases you have!
4. Let's see how we can improve…
5. What if my request fails?
6. Retrying requests Network requests on mobile can fail due to a multitude of reasons. Often, you'll want some sort of mechanism for retrying a request. getRequestObservable() .subscribe(response -> { // handle response });
7. Retrying requests - Naive approach The most simple way is to recreate your source Observable every time that you want to make a request. Since you're automatically retrying, you run the risk of overrunning your servers in error scenarios. private void makeRequest() { getRequestObservable() .subscribe(getObserver()); } private Observer<Response> getObserver() { return new Observer<>() { … @Override public void onError(Throwable t) { if (someCondition) { makeRequest(); } } } }
8. Retrying requests - retryWhen() RxJava has retryWhen()! retryWhen() can be a little tricky to grok. Let's walk through it: • retryWhen() is called once per subscription • It's given an Observable of the errors by the parent stream (Observable<Throwable>) • As long as the Observable returned from retryWhen() does not complete or error, the parent will be resubscribed // this code retries the request three times, with a 5s delay, // then 10s, then 15s getRequestObservable() .retryWhen(attempt -> { return attempt .zipWith(Observable.range(1, 3), (n, i) -> i) .flatMap(i -> return Observable.timer(5 * i, TimeUnit.SECONDS); }) }) .subscribe(viewModel -> { // handle updated request state });
9. Retrying requests - retryWhen() Some of you might be very concerned though, automatic retries are dangerous business! If you retry too often you can easily have your apps DDOS'ing your servers. So, let's only retry when the user wants to retry… Fortunately this is easy to do with Relays/Subjects PublishRelay<Long> retryRequest = PublishRelay.create(); getRequestObservable() .retryWhen(attempt -> retryRequest) .subscribe(viewModel -> { // handle updated response }, t -> { retryView.setVisibility(View.VISIBLE); }); @OnClick(R.id.retry_view) public void onRetryClicked() { retryRequest.call(System.currentTimeMillis); }
10. My API has very specific response codes…
11. Using Response<T> In our canonical example, we're using an Observable representing the typed body of the response document. This is not a requirement! If you need access to metadata about the response, you can use Observable<Response<T>> One nicety of this approach is that server responses in the 400-500 level will not cause an exception in your stream. interface SeatGeekApi { @GET("events") Observable<Response<EventsResponse>> getUpcomingEvents(); } Retrofit retrofit = new Retrofit.Builder() .baseUrl("https://api.seatgeek.com/") .build(); SeatGeekApi api = retrofit.create(SeatGeekApi.class); api.getUpcomingEvents() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(eventsResponse -> { int responseCode = eventsResponse.code(); switch (responseCode) { HTTP_301: … HTTP_403: … } }, error -> { // note: only handle i/o errors! });
12. Using Response<T> - Just errors Always using Response<T> can be cumbersome though if you only want to check error codes. Fortunately, Retrofit has you covered with HttpException. You can check your error in onError to see if you have an HttpException to work with. interface SeatGeekApi { @GET("events") Observable<EventsResponse> getUpcomingEvents(); } Retrofit retrofit = new Retrofit.Builder() .baseUrl("https://api.seatgeek.com/") .build(); SeatGeekApi api = retrofit.create(SeatGeekApi.class); api.getUpcomingEvents() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(eventsResponse -> { // do something with events }, error -> { if (error instanceof HttpException) { Response response = ((HttpException)error).response(); switch (response.code()) { … } } else { // handle other errors } });
13. Using Response<T> - With share() Sometimes to help with readability or because you have specific branching conditions downstream, it can be useful to multicast your request with share(). share() lets multiple subscribers handle the data provided from upstream, in our case this is the network response. In this example, we're able to set up dedicated subscribers for specific response codes. interface SeatGeekApi { @GET("events") Observable<Response<EventsResponse>> getUpcomingEvents(); } Observable<Response<EventsResponse>> eventsResponse = api.getUpcomingEvents() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .share(); eventsResponse .filter(Response::isSuccessful) .subscribe(this::handleSuccessfulResponse); eventsResponse .filter(response -> response.code() == HTTP_403) .subscribe(this::handle403Response); eventsResponse .filter(response -> response.code() == HTTP_304) .subscribe(this::handle304Response);
14. My designer wants a loading indicator…
15. Showing loading The goal with showing a loading state is that you'll want a simple way to show loading without tightly coupling your UI to the request stream First, we'll look at the naive approach... api.getUpcomingEvents() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(events -> { // do something with events }, error -> { // handle errors });
16. Showing loading - Naive approach What are the problems here? • We've tightly coupled our UI to the stream • These lambdas leak references to our Views (more on subscription management in a bit) • Can't handle "streams", only works as a single request api.getUpcomingEvents() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .doOnSubscribe(() -> loadingIndicator.show()) .doOnUnsubscribe(() -> loadingIndicator.hide()) .subscribe(events -> { // do something with events }, error -> { // handle errors });
17. Showing loading - Splitting streams So, let's try something new. What if our loading state was actually a separate stream? This gives us flexibility to have our UI subscribe to the loading state separately enum RequestState { IDLE, LOADING, COMPLETE, ERROR } BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE); Observable.just(trigger) .doOnNext(() -> state.call(RequestState.LOADING)) .observeOn(Schedulers.io()) .flatMap(trigger -> api.getUpcomingEvents()) .observeOn(AndroidSchedulers.mainThread()) .doOnError(t -> state.call(RequestState.ERROR)) .doOnComplete(() -> state.call(RequestState.COMPLETE)) .subscribe(events -> { // do something with events }, error -> { // handle errors });
18. Showing loading - Splitting streams Let's look at how the UI then subscribes to our request state then… We can now manage our view updates separately from the request stream! enum RequestState { IDLE, LOADING, COMPLETE, ERROR } BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE); state.subscribe(requestState -> { switch(requestState) { IDLE: break; LOADING: loadingIndicator.show(); errorView.hide(); break; COMPLETE: loadingIndicator.hide(); break; ERROR: loadingIndicator.hide(); errorView.show(); break; } );
19. Showing loading - Splitting streams We can extend this model then to other aspects of the request! Now we've decoupled the UI from the request stream completely, success! class ApiRequest { BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE); BehaviorRelay<EventsResponse> response = BehaviorRelay.create(); BehaviorRelay<Throwable> errors = BehaviorRelay.create(); BehaviorRelay<Long> trigger = BehaviorRelay.create(); public ApiRequest() { trigger .doOnNext(() -> state.call(RequestState.LOADING)) .observeOn(Schedulers.io()) .flatMap(trigger -> api.getUpcomingEvents()) .observeOn(AndroidSchedulers.mainThread()) .doOnError(t -> state.call(RequestState.ERROR)) .doOnError(errors) .onErrorResumeNext(Observable.empty()) .doOnNext(() -> state.call(RequestState.COMPLETE)) .subscribe(response); } void execute() { trigger.call(System.currentTimeMilis()); } }
20. Showing loading - Splitting streams Now our UI can subscribe independently to the data and errors and handle them separately as desired. ApiRequest request = new ApiRequest(api.getUpcomingEvents()); request.state.subscribe(requestState -> updateLoadingView(requestState)); request.state.subscribe(requestState -> updateLoadingNotification(requestState)); request.response.subscribe(eventsResponse -> showEventsResponse(eventsResponse)); request.errors.subscribe(throwable -> extractAndShowError(throwable)); request.execute();
21. Showing loading - Splitting streams You can always combine these streams together in the UI if you don't like handling to be spread out… BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE); BehaviorRelay<Optional<EventsResponse>> response = BehaviorRelay.create(Optional.absent()); BehaviorRelay<Optional<Throwable>> errors = BehaviorRelay.create(Optional.absent()); class RequestViewModel { public final RequestState state; public final Optional<EventsResponse> response; public final Optional<Throwable> errors; RequestViewModel(...) { … } } Observable.combineLatest(state, response, errors, RequestViewModel::new) .subscribe(viewModel -> { // handle updated request state });
22. Sealed Classes If you wanted to expand on the previous thought, you can utilize a Kotlin language feature that makes consolidating our request into a single stream just a little bit easier. sealed class RequestState { object Loading : RequestState() data class Complete(val response: EventsResponse) : RequestState() data class Error(val error: Throwable) : RequestState() } Observable.fromCallable({ api.getUpcomingEvents().execute() }) .subscribeOn(Schedulers.io()) .startWith(RequestState.Loading) .onErrorReturn { RequestState.Error(it) } .observeOn(AndroidSchedulers.mainThread()) .subscribe { requestState -> when (requestState) { is Idle -> clearView() is Loading -> showLoading() is Completed -> showData(requestState.response) is Error -> showError(requestState.error) } }
23. How about configuration changes?
24. Subscription management RxJava, like AsyncTask and other asynchronous patterns do not work with the Android lifecycles by default… RxJava provides a method to unsubscribe from streams via Disposable But the key question is how to use it properly? Disposable disposable = Observable .combineLatest(state, response, errors, RequestViewModel::new) .subscribe(viewModel -> { // handle updated request state }); ... disposable.dispose(); // when?
25. Subscription management - Basic Disposable Basic Disposable usage is hopefully straightforward. Keep a reference to the returned object from your subscription and then dispose() of it in an appropriate lifecycle method. class MyActivity extends Activity { Disposable disposable; … disposable = Observable .combineLatest(state, response, errors, ViewModel::new) .subscribe(getSubscriber()); … public void onStop() { disposable.dispose(); } }
26. Subscription management - CompositeDisposable CompositeDisposable is a very useful tool if you have your UI subscribed to multiple streams. You can add() Disposables to it so that you can have them all be disposed at the same time! Note: This can be very convenient in a MVP-like system where you want all your subscriptions to end in an unbind() method. class MyActivity extends Activity { CompositeDisposable disposables = new CompositeDisposable(); … disposables.add(state.subscribe(getStateSubscriber())); disposables.add(result.subscribe(getResultSubscriber())); disposables.add(error.subscribe(getErrorSubscriber())); … public void onStop() { disposable.clear(); } }
27. Subscription management - RxLifecycle https://github.com/trello/RxLifecycle RxLifecycle gives a simple way to tie subscriptions to your Activity/Fragment lifecycle. When a corresponding lifecycle event occurs, the Disposable will be dispose() 'd class MyActivity extends RxAppCompatActivity { … Observable .combineLatest(state, response, errors, ViewModel::new) .compose(bindToLifecycle()) .subscribe(viewModel -> { // handle updated request state }); … }
28. Subscription management - nonConfigurationInstance Out of the box, there's not an easy way to continue an Observable stream across configuration changes. However, we can coopt the Android Activity's onRetainNonConfigurationInstance() so that our stream is accessible again after a configuration change… Note: this is also a handy strategy for keeping your Presenter/Dagger graph around for your next Activity class MyActivity extends Activity { Observable<ViewModel> request; Disposable disposable; public void onCreate(Bundle instanceState) { request = getLastNonConfigurationInstance(); } … request = Observable .combineLatest(state, response, errors, ViewModel::new) .cache(); disposable = request .subscribe(getSubscriber()); … public void onStart() { if (request != null && disposable == null) { disposable = request .subscribe(getSubscriber()); // can null request field // in subscriber } } Object onRetainNonConfigurationInstance() { return request; } public void onStop() { disposable.dispose(); } }
29. Subscription management - RxLifecycle + retain If we want to retain our Observable, that works with RxLifecycle too. class MyActivity extends Activity { Observable<ViewModel> request; public void onCreate(Bundle instanceState) { request = getLastNonConfigurationInstance(); } … public void onStart() { if (request != null) { request .compose(bindToLifecycle()) .subscribe(getSubscriber()); } else { request = Observable .combineLatest(state, response, errors, ViewModel::new) .cache(); request .compose(bindToLifecycle()) .subscribe(getSubscriber()); } } Object onRetainNonConfigurationInstance() { return request; } public void onStop() { disposable.dispose(); } }
30. Subscription management - RxLifecycle + MVP You can customize your MVP Presenter as well to take advantage of similar functionality! Now, you can bind your observation of the request streams and never worry about your view being detached. enum PresenterLifecycle { STOPPED, STARTED } class RxPresenter<V extends View> extends Presenter<V> { BehaviorRelay<PresenterLifecycle> presenterLifecycle = BehaviorRelay.create(PresenterLifecycle.STOPPED); void bind(V view) { presenterLifecycle.call(PresenterLifecycle.STARTED); } void unbind() { presenterLifecycle.call(PresenterLifecycle.STOPPED); } protected <T> LifecycleTransformer<T> bindToLifecycle() { return RxLifecycle.bind(presenterLifecycleObservable, lifecycle -> { switch (lifecycle) { case STARTED: return PresenterLifecycle.STOPPED; case STOPPED: default: return PresenterLifecycle.STARTED; } }); } }
31. Questions?
32. Thanks! About me… name: Stephen D’Amico job: Android Lead, SeatGeek twitter: @sddamico github: sddamico