Creating custom rx.Observable and rx.Subsciber

Question

I'm working on an Android App. I'm using Retrofit to manage the http request to the server-side endpoints. Currently when I'm doing a request I'm doing something like this:

Observable<List<ApiFeedResponse>> feedObservable = mFeedRepository.getFeed(0, 50)
    .flatMap(apiFeedsResponse -> {
        if (apiFeedsResponse.code() != 200) {
            if (apiFeedsResponse.code() == 304) {
                List<ApiFeedResponse> body = apiFeedsResponse.body();
                return Observable.just(body);
            } else {
                return Observable.error(new ServerSideErrorException(apiFeedsResponse));
            }
        } else {
            return Observable.just(apiFeedsResponse.body());
        }
    });

My FeedRepository calls the Retrofit service. I've an endpoint that is like myhost.com/rest/userfeed?page=0&pageSize=50. The thing is that I'm also using etags to get cached server-side responses. And I want to be able to differentiate between a normal 200 http response and a "not modified" 304 response. I want to extend the rx.Subscriber lifecycle methods (onNext, onError and onComplete) to be something like (onSuccess, onServerError, onNotModified, onServerResult). That way when I subscribe to this methods is going to look like this:

getFeed(0, 50).subscribe(new ServerSubscriber<List<Feed>>() {

    //Executed when the response is 200
    @Override
    protected void onSuccess(List<Feed> feed) {

    }

    //Executed when the response is 304
    @Override
    protected void onNotModified(List<Feed> feed) {

    }

    //Executed if something goes wrong while doing the http request (code is different than 200 or 304)
    @Override
    protected void onServerSideError(ServerSideErrorException e) {

    }

    //Executed always that the result of the http request is successfull (200 or 304)
    @Override
    protected void onServerResult(List<Feed> feed) {

    }

});

I have been looking through different repos trying to find if someone has done something similar and the closes thing that I found was this: https://github.com/ReactiveX/RxJava/issues/1034

But I still can't fully understand how to implement custom rx.Observables and custom rx.Subscribers. Any advice is welcome.

Tassos Bassoukos · Accepted Answer · 2017-04-21 22:12:40Z

2

Why not repackage your logic in a reusable form?

public <T> Transformer<Response<T>, T> applyCache( 
                 Supplier<T> src,
                 Consumer<Response<T>> sink) {
    return responseSrc -> responseSrc.flatMap(response -> {
    switch(response.code()) {
    case 200:
        sink.accept(response);
        return Observable.just(apiFeedsResponse.body());
    case 304:
         return Observable.just(src.get());
    default:
         return Observable.error(new ServerSideErrorException(apiFeedsResponse));
    }
}

Just add the cache get/set functions (and adjust to taste); use like:

 mFeedRepository
 .getFeed(0, 50)
 .compose(applyCache(feedCache::get, feedCache::set)

answered Apr 21, 2017 at 22:12

Tassos Bassoukos

16.2k2 gold badges39 silver badges42 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

4gus71n Over a year ago

Yeah, this is what I was looking for. Do you know if there's any way to hook this directly using Retrofit? That way I can avoid setting the .compose line to every Repository method.

Tassos Bassoukos Over a year ago

Unfortunately there isn't; but then again that's better done at the OkHttp level by adding an interceptor.

4gus71n Over a year ago

BTW apparently it can be added in the Retrofit layer. Here's an example github.com/square/retrofit/pull/1865

Ajinkya Over a year ago

stackoverflow.com/q/46214639/2009899 can you help on this question.

LordRaydenMK · Accepted Answer · 2017-04-21 18:37:28Z

2

I think the solutions to your problems can be fixed in Retrofit/OkHttp.

Retrofit2 uses OkHttp3 under the hood to execute the API calls. OkHttp can handle the 304 not modified status code and deliver you the result from cache. To do that you need to set up retrofit to use a custom OkHttp client with cache.

For more custom callbacks there the solution is custom CallAdapter. There is an example for that in the retrofit repo which looks similar to yours. RxJava already uses a call adapter, maybe you can base is on that.

answered Apr 21, 2017 at 18:37

LordRaydenMK

13.3k5 gold badges53 silver badges56 bronze badges

1 Comment

4gus71n Over a year ago

Yeah I already enabled the OkHttp cache, the thing is that my Retrofit interfaces are returning rx.Observable<Response<SomeObject>>. I need to differentiate between 200 responses and 304 responses because some of my server API calls need to update a SQL database when It's a 200 response, but I don't need to update the database when It's a 304 response. The ErrorHandlingAdapter works if the retrofit interfaces uses Call<Response<SomeObject>>.

Collectives™ on Stack Overflow

Creating custom rx.Observable and rx.Subsciber

2 Answers 2

4 Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

4 Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related