Sync Recrods Between Local And Server using RxJava

Question

I am trying to sync some data between local records and the records on the server. syncList is the entry point for the sync operation with local records. Inside this method i call a method syncResultObservable that would return the Observable object with the records on the server.

Once the records from the server are downloaded I use flatmap to transform the sync result into a map<String, List<String>> by calling syncResult (List<String> local, List<String> server). The resultant map should have 2 keys "toAdd":List<String> "toRemove":List<String>

Now, i want to perform separate operations on the each result. recordsAddObservable should add the list from toAdd key (i am able to do this). recordsRemoveObservable should remove add the list from toRemove key (could not find a way to perform this operation using the same observable).

recordsAddObservable operation toAdd records is working fine for me. However, i am trying to find a way to call recordsRemoveObservable on the map result without causing the whole sync operation to start again for remove case. Below is the small snippet for 'toAdd' records.

public void syncList(final List<String> localData) {
   //fetch the server data
    Observable<Map<String, List<String>>> syncResultObservable = serverDataDownloadObservable(userid)
            .flatMap(new Func1<String[], Observable<Map<String, List<String>>>>() {
                @Override
                public Observable<Map<String, List<String>>> call(String[] serverData) {
                    Map<String, List<String>> map = syncResult(localData, Arrays.asList(serverData));
                    return Observable.just(map);
                }
            });

    syncResultObservable.flatMap(new Func1<Map<String, List<String>>, Observable<List<String>>>() {
        @Override
        public Observable<List<String>> call(Map<String, List<String>> stringListMap) {
            List<String> toAdd;
            toAdd = stringListMap.get("toAdd");
            if (toAdd == null || toAdd.isEmpty()) {
                toAdd = new ArrayList<String>();
            }
            return Observable.just(toAdd);
        }
    }).flatMap(new Func1<List<String>, Observable<Result>>() {
        @Override
        public Observable<Result> call(List<String> strings) {
            if (strings.isEmpty() == false) {
                String[] toAdd = strings.toArray(new String[strings.size()]);
                Log.i(LogEnum.LogPriority.info, getClass, "syncRecords: will try to add records ", toAdd);
                return apis.recordsAddObservable(userid, toAdd);
            }
                Log.i("Tag", getClass, "syncRecords: no records to add, toAdd);

            Log.i("Tag", getClass, "syncRecords: no records need to be added ");
            return Observable.just(new Result(null, null));
        }
    }).subscribeOn(Schedulers.io()).subscribe(new Subscriber<Result>() {
        @Override
        public void onCompleted() {
             //
        }

        @Override
        public void onError(Throwable e) {
        }

        @Override
        public void onNext(Result result) {
            onCompleted();
        }
    });
}






public Map<String, List<String>> syncResult (List<String> local, List<String>  server) {
    . . .
    . . .

    Map<String, List<String>> map = new HashMap<>();
    map.put("toAdd", toAddList);
    map.put("toRemove", toRemoveList);
    return map
}

Please let me know if there is a simple way to perform individual operation of different result sets.

Thanks Big O

Answer 1

Here is a runnable example which, I think, does what you want:

• take a stream of the local values and the server values,
• apply toList on them,
• zip them together and apply the difference calculation logic,
• since you chose to use a map with specific keys, we need to flatten it into keyed streams,
• then create groups based on the two keys,
• and switch over the keys in a flatMap to start the async work required by each case.