refactor sync of item state (chunk sync)

2 files changed, 86 insertions, 52 deletions

diff --git a/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/ItemStateSync.java b/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/ItemStateSync.java
index 284b0a37..a9e4f069 100644
--- a/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/ItemStateSync.java
+++ b/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/ItemStateSync.java
@@ -1,9 +1,18 @@
 package de.luhmer.owncloudnewsreader.reader.nextcloud;
 
+import android.util.Log;
+
 import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 import de.luhmer.owncloudnewsreader.database.DatabaseConnectionOrm;
+import de.luhmer.owncloudnewsreader.database.model.Feed;
 import de.luhmer.owncloudnewsreader.reader.FeedItemTags;
 
 /**
@@ -12,55 +21,88 @@ import de.luhmer.owncloudnewsreader.reader.FeedItemTags;
 
 public class ItemStateSync {
 
+    private static final String TAG = ItemStateSync.class.getCanonicalName();
+
     public static boolean PerformItemStateSync(NewsAPI newsApi, DatabaseConnectionOrm dbConn) throws IOException {
-        boolean successful;
-
-        //Mark as READ
-        List<String> itemIds = dbConn.getRssItemsIdsFromList(dbConn.getAllNewReadRssItems());
-        boolean result = PerformTagExecution(itemIds, FeedItemTags.MARK_ITEM_AS_READ, dbConn, newsApi);
-        if(result)
-            dbConn.change_readUnreadStateOfItem(itemIds, true);
-        successful = result;
-
-        //Mark as UNREAD
-        itemIds = dbConn.getRssItemsIdsFromList(dbConn.getAllNewUnreadRssItems());
-        result = PerformTagExecution(itemIds, FeedItemTags.MARK_ITEM_AS_UNREAD, dbConn, newsApi);
-        if(result)
-            dbConn.change_readUnreadStateOfItem(itemIds, false);
-        successful &= result;
-
-        //Mark as STARRED
-        itemIds = dbConn.getRssItemsIdsFromList(dbConn.getAllNewStarredRssItems());
-        result = PerformTagExecution(itemIds, FeedItemTags.MARK_ITEM_AS_STARRED, dbConn, newsApi);
-        if(result)
-            dbConn.changeStarrUnstarrStateOfItem(itemIds, true);
-        successful &= result;
-
-        //Mark as UNSTARRED
-        itemIds = dbConn.getRssItemsIdsFromList(dbConn.getAllNewUnstarredRssItems());
-        result = PerformTagExecution(itemIds, FeedItemTags.MARK_ITEM_AS_UNSTARRED, dbConn, newsApi);
-        if(result)
-            dbConn.changeStarrUnstarrStateOfItem(itemIds, false);
-        successful &= result;
+        boolean successful = true;
+
+        int MAX_SYNC_ITEMS_PER_REQUEST = 300;
+
+        Map<FeedItemTags, List<String>> itemsToSync = new HashMap<>();
+        itemsToSync.put(
+                FeedItemTags.MARK_ITEM_AS_READ,
+                dbConn.getRssItemsIdsFromList(dbConn.getAllNewReadRssItems())
+        );
+        itemsToSync.put(
+                FeedItemTags.MARK_ITEM_AS_UNREAD,
+                dbConn.getRssItemsIdsFromList(dbConn.getAllNewUnreadRssItems())
+        );
+        itemsToSync.put(
+                FeedItemTags.MARK_ITEM_AS_STARRED,
+                dbConn.getRssItemsIdsFromList(dbConn.getAllNewStarredRssItems())
+        );
+        itemsToSync.put(
+                FeedItemTags.MARK_ITEM_AS_UNSTARRED,
+                dbConn.getRssItemsIdsFromList(dbConn.getAllNewUnstarredRssItems())
+        );
+
+        Log.d(TAG, "itemsToSync[MARK_ITEM_AS_READ]:" + itemsToSync.get(FeedItemTags.MARK_ITEM_AS_READ).size());
+        Log.d(TAG, "itemsToSync[MARK_ITEM_AS_UNREAD]:" + itemsToSync.get(FeedItemTags.MARK_ITEM_AS_UNREAD).size());
+        Log.d(TAG, "itemsToSync[MARK_ITEM_AS_STARRED]:" + itemsToSync.get(FeedItemTags.MARK_ITEM_AS_STARRED).size());
+        Log.d(TAG, "itemsToSync[MARK_ITEM_AS_UNSTARRED]:" + itemsToSync.get(FeedItemTags.MARK_ITEM_AS_UNSTARRED).size());
+
+
+        for(Map.Entry<FeedItemTags, List<String>> entry : itemsToSync.entrySet()) {
+            FeedItemTags operation = entry.getKey();
+            Collection<List<String>> itemIdsPartitioned = partitionBasedOnSize(entry.getValue(), MAX_SYNC_ITEMS_PER_REQUEST);
+            for(List<String> itemIds : itemIdsPartitioned) {
+                Log.d(TAG, "Marking " + itemIds.size() + " items as " + operation.toString());
+                successful &= PerformTagExecution(itemIds, operation, dbConn, newsApi);
+            }
+        }
 
         return successful;
     }
 
+    static <T> Collection<List<T>> partitionBasedOnSize(List<T> inputList, int size) {
+        final AtomicInteger counter = new AtomicInteger(0);
+        return inputList.stream()
+                .collect(Collectors.groupingBy(s -> counter.getAndIncrement()/size))
+                .values();
+    }
+
     private static boolean PerformTagExecution(List<String> itemIds, FeedItemTags tag, DatabaseConnectionOrm dbConn, NewsAPI newsApi) throws IOException {
         if(itemIds.size() <= 0) { // Nothing to sync --> Skip
             return true;
         }
 
+        boolean success = false;
         switch(tag) {
             case MARK_ITEM_AS_READ:
-                return newsApi.markItemsRead(new ItemIds(itemIds)).execute().isSuccessful();
+                success = newsApi.markItemsRead(new ItemIds(itemIds)).execute().isSuccessful();
+                if (success) {
+                    dbConn.change_readUnreadStateOfItem(itemIds, true);
+                }
+                break;
             case MARK_ITEM_AS_UNREAD:
-                return newsApi.markItemsUnread(new ItemIds(itemIds)).execute().isSuccessful();
+                success = newsApi.markItemsUnread(new ItemIds(itemIds)).execute().isSuccessful();
+                if(success) {
+                    dbConn.change_readUnreadStateOfItem(itemIds, false);
+                }
+                break;
             case MARK_ITEM_AS_STARRED:
-                return newsApi.markItemsStarred(new ItemMap(itemIds, dbConn)).execute().isSuccessful();
+                success = newsApi.markItemsStarred(new ItemMap(itemIds, dbConn)).execute().isSuccessful();
+                if(success) {
+                    dbConn.changeStarrUnstarrStateOfItem(itemIds, true);
+                }
+                break;
             case MARK_ITEM_AS_UNSTARRED:
-                return newsApi.markItemsUnstarred(new ItemMap(itemIds, dbConn)).execute().isSuccessful();
+                success = newsApi.markItemsUnstarred(new ItemMap(itemIds, dbConn)).execute().isSuccessful();
+                if(success) {
+                    dbConn.changeStarrUnstarrStateOfItem(itemIds, false);
+                }
+                break;
         }
-        return false;
+        return success;
     }
 }
diff --git a/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/RssItemObservable.java b/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/RssItemObservable.java
index ff6abef2..0fd64bbd 100644
--- a/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/RssItemObservable.java
+++ b/News-Android-App/src/main/java/de/luhmer/owncloudnewsreader/reader/nextcloud/RssItemObservable.java
@@ -58,28 +58,22 @@ public class RssItemObservable implements Publisher<Integer> {
     }
 
     public void sync(Subscriber<? super Integer> subscriber) throws IOException {
-
         mDbConn.clearDatabaseOverSize();
 
-        //String authKey = AuthenticationManager.getGoogleAuthKey(username, password);
-        //int maxItemsInDatabase = Integer.parseInt(mPrefs.getString(SettingsActivity.SP_MAX_ITEMS_SYNC, "200"));
-
         long lastModified = mDbConn.getLastModified();
-        //dbConn.clearDatabaseOverSize();
 
         int requestCount = 0;
         int totalCount   = 0;
         int maxSyncSize = maxSizePerSync;
 
-        if(lastModified == 0)//Only on first sync
-        {
+        if (lastModified == 0) { // Only on first sync
             long offset = 0;
 
             Log.v(TAG, "First sync - download all available unread articles!!");
-            int maxItemsInDatabase = Constants.maxItemsCount;
+            // int maxItemsInDatabase = Constants.maxItemsCount;
 
             do {
-                Log.v(TAG, "offset=" + offset + ",  requestCount=" + requestCount + "");
+                Log.v(TAG, "[unread] offset=" + offset + ",  requestCount=" + requestCount + ", maxSyncSize=" + maxSyncSize + ", total downloaded=" + totalCount);
                 List<RssItem> buffer = (mNewsApi.items(maxSyncSize, offset, Integer.parseInt(FeedItemTags.ALL.toString()), 0, false, true).execute().body());
 
                 requestCount = 0;
@@ -110,38 +104,36 @@ public class RssItemObservable implements Publisher<Integer> {
                     offset = getMaxIdFromItems(buffer); // get maximum id of returned items
                     performDatabaseBatchInsert(mDbConn, buffer);
                 }
-                Log.v(TAG, "[starred] offset=" + offset + ",  requestCount=" + requestCount + ", maxSyncSize=" + maxSyncSize);
+                Log.v(TAG, "[starred] offset=" + offset + ",  requestCount=" + requestCount + ", maxSyncSize=" + maxSyncSize + ", total downloaded=" + totalCount);
                 totalCount += requestCount;
 
                 subscriber.onNext(totalCount);
             } while(requestCount == maxSyncSize);
-        }
-        else
-        {
+        } else {
             Log.v(TAG, "Incremental sync!!");
             //First reset the count of last updated items
             mPrefs.edit().putInt(Constants.LAST_UPDATE_NEW_ITEMS_COUNT_STRING, 0).apply();
 
-            //long highestItemIdBeforeSync = mDbConn.getHighestItemId();
+            // long highestItemIdBeforeSync = mDbConn.getHighestItemId();
 
-            //Get all updated items
+            // Get all updated items
             mNewsApi.updatedItems(lastModified, Integer.parseInt(FeedItemTags.ALL.toString()), 0)
                     .flatMap((Function<ResponseBody, ObservableSource<RssItem>>) responseBody -> events(responseBody.source()))
-                    .subscribe(new Observer<RssItem>() {
+                    .subscribe(new Observer<>() {
                         int totalUpdatedUnreadItemCount = 0;
                         final int bufferSize = 150;
                         final List<RssItem> buffer = new ArrayList<>(bufferSize); //Buffer of size X
 
                         @Override
                         public void onSubscribe(@NonNull Disposable d) {
-                            Log.v(TAG, "onSubscribe() called with Disposable: d = [" + d + "]");
+                            Log.v(TAG, "onSubscribe() called");
                         }
 
                         @Override
                         public void onNext(@NonNull RssItem rssItem) {
                             long rssLastModified = rssItem.getLastModified().getTime();
                             // If updated item is unread and last modification was different from last sync time
-                            if(!rssItem.getRead() && rssLastModified != lastModified) {
+                            if (!rssItem.getRead() && rssLastModified != lastModified) {
                                 totalUpdatedUnreadItemCount++;
                             }