Rxjava2 + Retrofit2 + Android. Лучший способ сделать сотни сетевых вызовов

Question

У меня есть приложение. У меня есть большая кнопка, которая позволяет пользователю синхронизировать все свои данные сразу с облаком. Функция повторной синхронизации, которая позволяет им отправлять все свои данные снова. (300+ записей)

Я использую RXjava2 и retrofit2. У меня мой модульный тест работает с одним вызовом. Однако мне нужно сделать N сетевых вызовов.

То, что я хочу избежать, имеет наблюдаемый вызов следующего элемента в очереди. Я нахожусь в точке, где мне нужно реализовать мою работоспособность. Я видел немного о Картах, но я не видел, чтобы кто-то использовал его в качестве очереди. Также я хочу избежать сбоя одного элемента, и он отчитывается, когда все элементы не работают, как функция Zip. Должен ли я просто делать неприятный класс менеджера, который отслеживает очередь? Или есть более чистый способ отправить несколько сотен предметов?

ПРИМЕЧАНИЕ: РЕШЕНИЕ НЕ МОЖЕТ ЗАВИСИТЬ В JAVA8/LAMBDAS. Это оказалось намного больше, чем оправдано.

Обратите внимание, что все предметы одного и того же объекта.

@Test 
public void test_Upload() { 
    TestSubscriber<Record> testSubscriber = new TestSubscriber<>(); 
    ClientSecureDataToolKit clientSecureDataToolKit = ClientSecureDataToolKit.getClientSecureDataKit(); 
    clientSecureDataToolKit.putUserDataToSDK(mPayloadSecureDataToolKit).subscribe(testSubscriber); 

    testSubscriber.awaitTerminalEvent(); 
    testSubscriber.assertNoErrors(); 
    testSubscriber.assertValueCount(1); 
    testSubscriber.assertCompleted(); 
} 

Мой помощник, чтобы собрать и отправить все мои пункты

public class SecureDataToolKitHelper { 
private final static String TAG = "SecureDataToolKitHelper"; 
private final static SimpleDateFormat timeStampSimpleDateFormat = 
     new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); 


public static void uploadAll(Context context, RuntimeExceptionDao<EventModel, UUID> eventDao) { 
    List<EventModel> eventModels = eventDao.queryForAll(); 

    QueryBuilder<EventModel, UUID> eventsQuery = eventDao.queryBuilder(); 
    String[] columns = {...}; 

    eventsQuery.selectColumns(columns); 

    try { 
     List<EventModel> models; 

     models = eventsQuery.orderBy("timeStamp", false).query(); 
     if (models == null || models.size() == 0) { 
      return; 
     } 

     ArrayList<PayloadSecureDataToolKit> toSendList = new ArrayList<>(); 
     for (EventModel eventModel : models) { 
      try { 
       PayloadSecureDataToolKit payloadSecureDataToolKit = new PayloadSecureDataToolKit(); 

       if (eventModel != null) { 


        // map my items ... not shown 

        toSendList.add(payloadSecureDataToolKit); 
       } 
      } catch (Exception e) { 
       Log.e(TAG, "Error adding payload! " + e + " ..... Skipping entry"); 
      } 
     } 

     doAllNetworkCalls(toSendList); 

    } catch (SQLException e) { 
     e.printStackTrace(); 
    } 

} 

мой дооснащения материал

public class ClientSecureDataToolKit { 

    private static ClientSecureDataToolKit mClientSecureDataToolKit; 
    private static Retrofit mRetrofit; 

    private ClientSecureDataToolKit(){ 
     mRetrofit = new Retrofit.Builder() 
     .baseUrl(Utilities.getSecureDataToolkitURL()) 
     .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) 
     .addConverterFactory(GsonConverterFactory.create()) 
     .build(); 
    } 

    public static ClientSecureDataToolKit getClientSecureDataKit(){ 
     if(mClientSecureDataToolKit == null){ 
      mClientSecureDataToolKit = new ClientSecureDataToolKit(); 
     } 
     return mClientSecureDataToolKit; 
    } 

    public Observable<Record> putUserDataToSDK(PayloadSecureDataToolKit payloadSecureDataToolKit){ 
     InterfaceSecureDataToolKit interfaceSecureDataToolKit = mRetrofit.create(InterfaceSecureDataToolKit.class); 
     Observable<Record> observable = interfaceSecureDataToolKit.putRecord(NetworkUtils.SECURE_DATA_TOOL_KIT_AUTH, payloadSecureDataToolKit); 
     return observable; 
    } 

} 

public interface InterfaceSecureDataToolKit { 

@Headers({ 
     "Content-Type: application/json" 
}) 

@POST("/api/create") 
Observable<Record> putRecord(@Query("api_token") String api_token, @Body PayloadSecureDataToolKit payloadSecureDataToolKit); 
} 

---

Update. Я пытаюсь применить этот ответ, чтобы не повезло. Сегодня у меня иссякает. Я пытаюсь осуществить это в качестве модульного тестирования, как я сделал для первоначального вызова для одного элемента .. Похоже, что-то не в порядке с использованием лямбда возможно ..

public class RxJavaBatchTest { 
    Context context; 
    final static List<EventModel> models = new ArrayList<>(); 

    @Before 
    public void before() throws Exception { 
     context = new MockContext(); 
     EventModel eventModel = new EventModel(); 
     //manually set all my eventmodel data here.. not shown 

     eventModel.setSampleId("SAMPLE0"); 
     models.add(eventModel); 
     eventModel.setSampleId("SAMPLE1"); 
     models.add(eventModel); 
     eventModel.setSampleId("SAMPLE3"); 
     models.add(eventModel); 


    } 

    @Test 
    public void testSetupData() { 
     Assert.assertEquals(3, models.size()); 
    } 

    @Test 
    public void testBatchSDK_Upload() { 


     Callable<List<EventModel> > callable = new Callable<List<EventModel> >() { 

      @Override 
      public List<EventModel> call() throws Exception { 
       return models; 
      } 
     }; 

     Observable.fromCallable(callable) 
       .flatMapIterable(models -> models) 
       .flatMap(eventModel -> { 
        PayloadSecureDataToolKit payloadSecureDataToolKit = new PayloadSecureDataToolKit(eventModel); 
        return doNetworkCall(payloadSecureDataToolKit) // I assume this is just my normal network call.. I am getting incompatibility errors when I apply a testsubscriber... 
          .subscribeOn(Schedulers.io()); 
       }, true, 1); 
    } 

    private Observable<Record> doNetworkCall(PayloadSecureDataToolKit payloadSecureDataToolKit) { 

     ClientSecureDataToolKit clientSecureDataToolKit = ClientSecureDataToolKit.getClientSecureDataKit(); 
     Observable observable = clientSecureDataToolKit.putUserDataToSDK(payloadSecureDataToolKit);//.subscribe((Observer<? super Record>) testSubscriber); 
     return observable; 
    } 

Результат ..

An exception has occurred in the compiler (1.8.0_112-release). Please file a bug against the Java compiler via the Java bug reporting page (http://bugreport.java.com) after checking the Bug Database (http://bugs.java.com) for duplicates. Include your program and the following diagnostic in your report. Thank you. 
com.sun.tools.javac.code.Symbol$CompletionFailure: class file for java.lang.invoke.MethodType not found 


FAILURE: Build failed with an exception. 

* What went wrong: 
Execution failed for task ':app:compile<MyBuildFlavorhere>UnitTestJavaWithJavac'. 
> Compilation failed; see the compiler error output for details. 

---

Редактировать. Больше не пытайтесь Лямбда. Даже после настройки пути на моем mac, javahome, чтобы указать 1,8 и т. Д., Я не мог заставить его работать. Если бы это был новый проект, я бы стал сильнее. Однако, поскольку это унаследованное приложение для Android, написанное веб-разработчиками, пытающимися андроид, это просто не отличный вариант. Не стоит тратить время на то, чтобы заставить его работать. Уже во дни этого задания вместо полудня он должен был принять.

Я не смог найти хороший пример без лямбда плоской карты. Я пробовал это сам, и это становилось грязным.

Answer 1

Если вы правильно поняли, что хотите сделать свои звонки параллельно?

Так гй-й способ сделать это было бы что-то вроде:

Observable.fromCallable(() -> eventsQuery.orderBy("timeStamp", false).query()) 
      .flatMapIterable(models -> models) 
      .flatMap(model -> { 
       // map your model 

       //avoid throwing exceptions in a chain, just return Observable.error(e) if you really need to 
       //try to wrap your methods that throw exceptions in an Observable via Observable.fromCallable() 


       return doNetworkCall(someParameter) 
         .subscribeOn(Schedulers.io()); 
      }, true /*because you don't want to terminate a stream if error occurs*/, maxConcurrent /* specify number of concurrent calls, typically available processors + 1 */) 
      .subscribe(result -> {/* handle result */}, error -> {/* handle error */}); 

В вашем ClientSecureDataToolKit переместить эту часть в конструктор

InterfaceSecureDataToolKit interfaceSecureDataToolKit = mRetrofit.create(InterfaceSecureDataToolKit.class);