Repeat A Retrofit Observable Request

Question

Я пытаюсь сделать запрос, который извлекает данные с сервера до тех пор, пока это не потянет данные. Проблема в том, что ответ имеет 21Data perPage. Но есть метатег, чтобы узнать, есть ли следующая страница. Итак, я могу тянуть до nextPage == totalPage.

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

Мой запрос Журналы

E/ServiceGen: http://theUrl.net/ 
    D/OkHttp: --> GET http://theUrl.net/lga?page=1&per_page=21 http/1.1 
    D/OkHttp: x-mobile: true 
    D/OkHttp: --> END GET 
    D/OkHttp: --> GET http://theUrl.net/lga?page=1&per_page=21 http/1.1 
    D/OkHttp: x-mobile: true 
    D/OkHttp: --> END GET 
    D/OkHttp: <-- 200 OK http://theUrl.net/lga?page=1&per_page=21 (929ms) 
    D/OkHttp: Date: Wed, 10 Aug 2016 09:01:00 GMT 
    D/OkHttp: Content-Type: application/json; charset=utf-8 
    D/OkHttp: <-- 200 OK http://theUrl.net/lga?page=1&per_page=21 (933ms) 
    D/OkHttp: Date: Wed, 10 Aug 2016 09:01:00 GMT 
    D/OkHttp: Content-Type: application/json; charset=utf-8 
    D/OkHttp: --> GET http://theUrl.net/lga?page=2&per_page=21 http/1.1 
    D/OkHttp: --> END GET 
    D/OkHttp: --> GET http://theUrl.net/lga?page=2&per_page=21 http/1.1 
    D/OkHttp: --> END GET 
    D/OkHttp: --> GET http://theUrl.net/lga?page=2&per_page=21 http/1.1 
    D/OkHttp: --> END GET 
    D/OkHttp: <-- 400 Bad Request http://theUrl.net/lga?page=2&per_page=21 (695ms) 
    D/OkHttp: <-- END HTTP (177-byte body) 
    D/OkHttp: <-- 400 Bad Request http://theUrl.net/lga?page=2&per_page=21 (696ms) 
    D/OkHttp: <-- END HTTP (177-byte body) 
    D/OkHttp: <-- 400 Bad Request http://theUrl.net/lga?page=2&per_page=21 (696ms) 

Если вы заметили http://theUrl.net/lga?page=1&per_page=21 называли дважды и http://theUrl.net/lga?page=3&per_page=21 был назван в 3 раза.

Итак, я решил использовать свой старый файл RestClient Class. И все получилось отлично. Но нет ничего плохого. Он запросил весь запрос до последнего. Я до сих пор не могу найти то, что случилось с моим ServiceGenerator class

RestClient класса File

public class RestClient { 

     private static final String TAG = "RestClient"; 
     private static ApiService apiEndpointInterface; 
     private static Context context; 

     /*static { 
      setupRestClient(); 
     }*/ 

     public static ApiService get(Context cont) { 
      context = cont; 
      if (apiEndpointInterface != null) 
       return apiEndpointInterface; 

      setupRestClient(); 
      return apiEndpointInterface; 
     } 

     private static void setupRestClient() { 
      // Define the interceptor, add authentication headers 
      Interceptor interceptor = chain -> { 
       Request newRequest = chain.request().newBuilder() 
         /*.addHeader("x-mobile", "true")*/ 
         .addHeader("Authorization", "Bearer " + PrefUtils.getToken(context)).build(); 
       return chain.proceed(newRequest); 
      }; 

      // Add the interceptor to OkHttpClient 
      OkHttpClient.Builder builder = new OkHttpClient.Builder(); 
      builder.interceptors().add(interceptor); 
      builder.addInterceptor(new HttpLoggingInterceptor().setLevel(HttpLoggingInterceptor.Level.BODY)); 
      builder.addNetworkInterceptor(new StethoInterceptor()); 
      builder.connectTimeout(30000, TimeUnit.SECONDS); 
      builder.readTimeout(30000, TimeUnit.SECONDS); 
      OkHttpClient client = builder.build(); 

      Retrofit retrofit = new Retrofit.Builder() 
        .baseUrl(BuildConfig.HOST) 
        .addCallAdapterFactory(RxJavaCallAdapterFactory.createWithScheduler(Schedulers.io())) 
        .addConverterFactory(GsonConverterFactory.create(CustomGsonParser.returnCustomParser())) 
        .client(client) 
        .build(); 
      apiEndpointInterface = retrofit.create(ApiService.class); 
     } 

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

Answer 1

Во-первых, почему вы создаете новый клиент и сервис для всех и каждого вызова? Создайте их один раз, сохраните их и повторно используйте.

Во-вторых, я бы сказал, что вы должны использовать BehaviorSubject (извините за Java 8, но это делает логика гораздо более заметным):

BehaviorSubject<Integer> subject = new BehaviorSubject<>(); 
Observable<T> obs = 
    subject 
    .flatMap(page -> 
     getPage(page) 
     .doOnNext(result -> { 
      if(result has next page) subject.onNext(page+1); 
      else      subject.onComplete(); 
     }), 1) 
    ; 

Теперь вы можете взять obs и извлекать объекты и делать что ты хочешь.

Edit: пост-комментарий, я бы сказал, попробовать что-то вроде этого:

public static Observable<LgaListResponse> getPages(Context acontext) { 
    String token = PrefUtils.getToken(acontext); 
    BehaviorSubject<Integer> pageControl = BehaviorSubject.<Integer>create(1); 
    return pageControl.concatMap(integer -> { 
     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.onComplete(); 
        } 
       }); 
    }).cache(); 
} 

Имейте в виду, что вы должны использовать getPages() один раз, и возвращает тот же Observable каждый раз для каждого контекста; cache() может обрабатывать несколько подписчиков и отменять подписку.