android - Retrofit2 Tail Recursion Using RxJava / RxAndroid

Question

I am really trying to get a hang of using Retrofit with RxJava / RxAndroid. I've done this using normal Retrofit2 Callback method in a previous app without the use of Reactive Programming and it worked fine. So, here is it. I need to Tail Recall a function meant to fetch all Local Government from the server. The API uses pagination (I have to construct the URL with ?page=1, perPage=2). I've to do this till I've the whole data. So, below is my Rx code

    public static Observable<LgaListResponse> getPages(Context acontext) {
    String token = PrefUtils.getToken(acontext);
    BehaviorSubject<Integer> pageControl = BehaviorSubject.<Integer>create(1);
    Observable<LgaListResponse> ret2 = pageControl.asObservable().concatMap(integer -> {
        if (integer > 0) {
            Log.e(TAG, "Integer: " + integer);
            return ServiceGenerator.createService(ApiService.class, token)
                    .getLgas(String.valueOf(integer), String.valueOf(21))
                    .doOnNext(lgaListResponse -> {
                        if (lgaListResponse.getMeta().getPage() != lgaListResponse.getMeta().getPageCount()) {
                            pageControl.onNext(initialPage + 1);
                        } else {
                            pageControl.onNext(-1);
                        }
                    });
        } else {
            return Observable.<LgaListResponse>empty().doOnCompleted(pageControl::onCompleted);
        }
    });

    return Observable.defer(() -> ret2);
}

And my ServiceGenerator Class

    public class ServiceGenerator {

        private static final String TAG = "ServiceGen";
        private static OkHttpClient.Builder builder = new OkHttpClient.Builder();

        private static Retrofit.Builder retrofitBuilder =
                new Retrofit.Builder()
                        .baseUrl(BuildConfig.HOST)
                        .addCallAdapterFactory(RxJavaCallAdapterFactory.createWithScheduler(Schedulers.io()))
                        .addConverterFactory(GsonConverterFactory.create(CustomGsonParser.returnCustomParser()));

        public static <S> S createService(Class<S> serviceClass, String token) {

            builder.addInterceptor(new HttpLoggingInterceptor().setLevel(HttpLoggingInterceptor.Level.BODY));
            /*builder.addNetworkInterceptor(new StethoInterceptor());*/
            builder.connectTimeout(30000, TimeUnit.SECONDS);
            builder.readTimeout(30000, TimeUnit.SECONDS);
            if (token != null) {
                Interceptor interceptor = chain -> {
                    Request newRequest = chain.request().newBuilder()
                            .addHeader("x-mobile", "true")
                            .addHeader("Authorization", "Bearer " + token).build();
                    return chain.proceed(newRequest);
                };
                builder.addInterceptor(interceptor);
            }
            OkHttpClient client = builder.build();

            Retrofit retrofit = retrofitBuilder.client(client).build();
            Log.e(TAG, retrofit.baseUrl().toString());
            return retrofit.create(serviceClass);
        }

        public static Retrofit retrofit() {
            OkHttpClient client = builder.build();
            return retrofitBuilder.client(client).build();
        }

        public static class CustomGsonParser {

            public static Gson returnCustomParser(){
                return new GsonBuilder()
                        .setExclusionStrategies(new ExclusionStrategy() {
                            @Override
                            public boolean shouldSkipField(FieldAttributes f) {
                                return f.getDeclaringClass().equals(RealmObject.class);
                            }

                            @Override
                            public boolean shouldSkipClass(Class<?> clazz) {
                                return false;
                            }
                        })
                        .create();
            }
        }
    }

So, I noticed on the first call, I get a response, but on the second one, the 440Error is thrown. The URL is formed, but the request throws a 400Error. I don't know why it's throwing a 400 everything is working fine if I use POSTMAN to test. And, I tested with my old code too. The Log is too long, so I put it in pastebin LOGS any help thanks. I've written most of this app with RxAndroid / RxJava. Thanks

Answer 1

I suggest you simplify things (and remove recursion). First build up your pages using something like

public static Observable<LgaListResponse> getPages(Context acontext, int initialPage, int perPage) {  
    String token = PrefUtils.getToken(acontext);
    BehaviorSubject<Integer> pagecontrol = BehaviorSubject.<Integer>create(initialPage);
    Observable<LgaListResponse> ret2 = pagecontrol.asObservable().concatMap(
        new Func1<Integer,Observable<LgaListResponse>>() {
            Observable<LgaListResponse> call(Integer pageNumber) {
                if (pageNumber > 0) {
                    return ServiceGenerator.createService(ApiService.class, token)
                                           .getLgas(String.valueOf(aKey), String.valueOf(perPage))
                                           .doOnNext(
                                                new Action1<LgaListResponse>() {
                                                    void call(LgaListResponse page) {
                                                        if (page.getMeta().getPage() != page.getMeta().getPageCount()) {
                                                            pagecontrol.onNext(page.getMeta().getNextPage());
                                                        } else {
                                                            pagecontrol.onNext(-1);
                                                        }         
                                                    }
                                                }
                                            );
                }                                               
                else {
                    return Observable.<LgaListResponse>empty().doOnCompleted(()->pagecontrol.onCompleted());
                }           
            }
        }
    );

    return Observable.defer(
                new Func0<Observable<LgaListResponse>() {
                    Observable<LgaListResponse> call() {
                        return ret2;
                    }
                }
            );            
}

then subscribe to the resulting observable. It looks horrible because I've avoided using lambdas but it should work.