Rxjava2 + Retrofit2 + Android. The best way to make hundreds of network calls

Question

I have an application. I have a big button that allows the user to synchronize all their data at once with the cloud. A resync feature that allows them to send all their data again. (300+ entries)

I am using RXjava2 and retrofit2. I have a unit test working with one call. However, I need to make N network calls.

What I want to avoid is to have an observable call to the next element in the queue. I am at the point where I need to realize my performance. I saw a little about Maps, but I did not see anyone using it as a queue. I also want to avoid the failure of one element, and it reports when all the elements do not work, like the Zip function. Should I just make an unpleasant manager class that tracks the queue? Or is there a cleaner way to ship a few hundred items?

NOTE. SOLUTION CANNOT DEPEND ON JAVA8 / LAMBDAS. This turned out to be much more justified.

Note that all elements are the same object.

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

My assistant to collect and send all my items

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

}

My retrofit

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. , . . unit test, . , - , .

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

, ?

, rx-y :

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