Refactor removeWatchedStreams() in LocalPlaylistFragment.

app/src/main/java/org/schabi/newpipe/local/playlist/LocalPlaylistFragment.java

@@ -11,6 +11,7 @@ import android.os.Parcelable;
 import android.text.InputType;
 import android.text.TextUtils;
 import android.util.Log;
+import android.util.Pair;
 import android.view.LayoutInflater;
 import android.view.Menu;
 import android.view.MenuInflater;
@@ -34,7 +35,6 @@ import org.schabi.newpipe.database.LocalItem;
 import org.schabi.newpipe.database.history.model.StreamHistoryEntry;
 import org.schabi.newpipe.database.playlist.PlaylistStreamEntry;
 import org.schabi.newpipe.database.stream.model.StreamEntity;
-import org.schabi.newpipe.database.stream.model.StreamStateEntity;
 import org.schabi.newpipe.databinding.DialogEditTextBinding;
 import org.schabi.newpipe.databinding.LocalPlaylistHeaderBinding;
 import org.schabi.newpipe.databinding.PlaylistControlBinding;
@@ -55,7 +55,6 @@ import org.schabi.newpipe.util.external_communication.ShareUtils;
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -63,7 +62,6 @@ import java.util.stream.Collectors;
 
 import icepick.State;
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
-import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
 import io.reactivex.rxjava3.disposables.Disposable;
@@ -309,7 +307,7 @@ public class LocalPlaylistFragment extends BaseLocalListFragment<List<PlaylistSt
     ///////////////////////////////////////////////////////////////////////////
 
     private Subscriber<List<PlaylistStreamEntry>> getPlaylistObserver() {
-        return new Subscriber<List<PlaylistStreamEntry>>() {
+        return new Subscriber<>() {
             @Override
             public void onSubscribe(final Subscription s) {
                 showLoading();
@@ -395,31 +393,19 @@ public class LocalPlaylistFragment extends BaseLocalListFragment<List<PlaylistSt
         isRemovingWatched = true;
         showLoading();
 
-        disposables.add(playlistManager.getPlaylistStreams(playlistId)
+        final var recordManager = new HistoryRecordManager(getContext());
-                .subscribeOn(Schedulers.io())
+        final var historyIdsFlowable = recordManager.getStreamHistorySortedById()
-                .map((List<PlaylistStreamEntry> playlist) -> {
+                // already sorted by ^ getStreamHistorySortedById(), binary search can be used
-                    // Playlist data
+                .map(historyList -> historyList.stream().map(StreamHistoryEntry::getStreamId)
-                    final Iterator<PlaylistStreamEntry> playlistIter = playlist.iterator();
+                        .collect(Collectors.toList()));
-
+        final var streamsFlowable = playlistManager.getPlaylistStreams(playlistId)
-                    // History data
+                .zipWith(historyIdsFlowable, (playlist, historyStreamIds) -> {
-                    final HistoryRecordManager recordManager =
-                            new HistoryRecordManager(getContext());
-                    final Iterator<StreamHistoryEntry> historyIter = recordManager
-                            .getStreamHistorySortedById().blockingFirst().iterator();
-
                     // Remove Watched, Functionality data
                     final List<PlaylistStreamEntry> notWatchedItems = new ArrayList<>();
                     boolean thumbnailVideoRemoved = false;
 
-                    // already sorted by ^ getStreamHistorySortedById(), binary search can be used
-                    final ArrayList<Long> historyStreamIds = new ArrayList<>();
-                    while (historyIter.hasNext()) {
-                        historyStreamIds.add(historyIter.next().getStreamId());
-                    }
-
                     if (removePartiallyWatched) {
-                        while (playlistIter.hasNext()) {
+                        for (final var playlistItem : playlist) {
-                            final PlaylistStreamEntry playlistItem = playlistIter.next();
                             final int indexInHistory = Collections.binarySearch(historyStreamIds,
                                     playlistItem.getStreamId());
 
@@ -432,14 +418,15 @@ public class LocalPlaylistFragment extends BaseLocalListFragment<List<PlaylistSt
                             }
                         }
                     } else {
-                        final Iterator<StreamStateEntity> streamStatesIter = recordManager
+                        final var streamStates = recordManager.loadLocalStreamStateBatch(playlist)
-                                .loadLocalStreamStateBatch(playlist).blockingGet().iterator();
+                                .blockingGet();
+
+                        for (int i = 0; i < playlist.size(); i++) {
+                            final var playlistItem = playlist.get(i);
+                            final var streamStateEntity = streamStates.get(i);
 
-                        while (playlistIter.hasNext()) {
-                            final PlaylistStreamEntry playlistItem = playlistIter.next();
                             final int indexInHistory = Collections.binarySearch(historyStreamIds,
                                     playlistItem.getStreamId());
-                            final StreamStateEntity streamStateEntity = streamStatesIter.next();
                             final long duration = playlistItem.toStreamInfoItem().getDuration();
 
                             if (indexInHistory < 0 || (streamStateEntity != null
@@ -453,19 +440,18 @@ public class LocalPlaylistFragment extends BaseLocalListFragment<List<PlaylistSt
                         }
                     }
 
-                    return Flowable.just(notWatchedItems, thumbnailVideoRemoved);
+                    return new Pair<>(notWatchedItems, thumbnailVideoRemoved);
-                })
+                });
+        disposables.add(streamsFlowable.subscribeOn(Schedulers.io())
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(flow -> {
-                    final List<PlaylistStreamEntry> notWatchedItems =
+                    final List<PlaylistStreamEntry> notWatchedItems = flow.first;
-                            (List<PlaylistStreamEntry>) flow.blockingFirst();
+                    final boolean thumbnailVideoRemoved = flow.second;
-                    final boolean thumbnailVideoRemoved = (Boolean) flow.blockingLast();
 
                     itemListAdapter.clearStreamItemList();
                     itemListAdapter.addItems(notWatchedItems);
                     saveChanges();
 
-
                     if (thumbnailVideoRemoved) {
                         updateThumbnailUrl();
                     }