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 });
fail • Dealing with specific response codes • Showing loading indicators in your UI • Configuration changes/subscription management • Probably other cases you have
a multitude of reasons. Often, you'll want some sort of mechanism for retrying a request. getRequestObservable() .subscribe(response -> { // handle response });
to recreate your source Observable every time that you want to make a request. private void makeRequest() { getRequestObservable() .subscribe(getObserver()); } private Observer<Response> getObserver() { return new Observer<>() { … @Override public void onError(Throwable t) { if (someCondition) { makeRequest(); } } } }
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 });
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 request state }); @OnClick(R.id.retry_view) public void onRetryClicked() { retryRequest.call(System.currentTimeMillis); }
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! });
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 } });
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);
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 });
• 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 });
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); void publishRequestState(RequestState requestState) { Observable.just(requestState) .observeOn(AndroidSchedulers.mainThread()) // ensure state updates // are on proper thread .subscribe(state); } api.getUpcomingEvents() .doOnSubscribe(() -> publishRequestState(RequestState.LOADING)) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .doOnError(t -> publishRequestState(RequestState.ERROR)) .doOnComplete(() -> publishRequestState(RequestState.COMPLETE)) .subscribe(events -> { // do something with events }, error -> { // handle errors });
then to other aspects of the request! Now we've decoupled the UI from the request stream completely, success! BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE); BehaviorRelay<EventsResponse> response = BehaviorRelay.create(); BehaviorRelay<Throwable> errors = BehaviorRelay.create(); … void executeRequest() { api.getUpcomingEvents() .doOnSubscribe(() -> publishRequestState(RequestState.LOADING)) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .doOnError(t -> publishRequestState(RequestState.ERROR)) .doOnComplete(() -> publishRequestState(RequestState.COMPLETE)) .subscribe(response, errors); } …
independently to the data and errors and handle them separately as desired. BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE); BehaviorRelay<EventsResponse> response = BehaviorRelay.create(); BehaviorRelay<Throwable> errors = BehaviorRelay.create(); state.subscribe(requestState -> updateLoadingView(requestState)); response.subscribe(eventsResponse -> showEventsResponse(eventsResponse)); errors.subscribe(throwable -> extractAndShowError(throwable));
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 });
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?
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(); } }
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(); } }
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(); } }
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 }); … }
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; } }); } }
sddamico
igreenwood
mkmk884
rainerhahnekamp
alexanderschranz
andpad
onopon
ma_cho29
seike460
lamodatech
maroon1st
manfredsteyer
fabpot
panda_program
kastner
nonsquared
brad_frost
lara
jacobian
shpigford
marcduiker
danielanewman
kevinliebholz
addyosmani
aarron
