Retrofit2 尾递归使用 RxJava / RxAndroid

Question

我真的想掌握使用Retrofit 和RxJava / RxAndroid 的窍门。我在以前的应用程序中使用普通的Retrofit2 Callback 方法完成了此操作，而没有使用Reactive Programming，并且效果很好。所以，这就是它。我需要Tail Recall 一个旨在从服务器获取所有Local Government 的函数。 API 使用分页 (I have to construct the URL with ?page=1, perPage=2)。我必须这样做，直到我拥有全部数据。所以，下面是我的Rx 代码

    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);
}

还有我的 ServiceGenerator 类

    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();
            }
        }
    }

所以，我注意到在第一次调用时，我得到了响应，但在第二次调用中，440Error 被抛出。 URL 已形成，但请求会引发 400Error。如果我使用POSTMAN 进行测试，我不知道为什么它会抛出400 一切正常。而且，我也用我的旧代码进行了测试。 Log 太长了，所以我把它放在pastebin LOGS 任何帮助谢谢。我用RxAndroid / RxJava 编写了这个应用程序的大部分内容。谢谢

Answer 1

我建议您简化事情（并删除递归）。首先使用类似的东西构建你的页面

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;
                    }
                }
            );            
}

然后订阅生成的 observable。它看起来很糟糕，因为我避免使用 lambda，但它应该可以工作。