No more reaction to unnecessary feed db-changes

This caused duplicate events (https://github.com/TeamNewPipe/NewPipe/pull/6686#issuecomment-909575283) and unnecessary processing of items

app/src/main/java/org/schabi/newpipe/database/feed/dao/FeedDAO.kt

@@ -7,6 +7,7 @@ import androidx.room.Query
 import androidx.room.Transaction
 import androidx.room.Update
 import io.reactivex.rxjava3.core.Flowable
+import io.reactivex.rxjava3.core.Maybe
 import org.schabi.newpipe.database.feed.model.FeedEntity
 import org.schabi.newpipe.database.feed.model.FeedLastUpdatedEntity
 import org.schabi.newpipe.database.stream.StreamWithState
@@ -37,7 +38,7 @@ abstract class FeedDAO {
         LIMIT 500
         """
     )
-    abstract fun getAllStreams(): Flowable<List<StreamWithState>>
+    abstract fun getAllStreams(): Maybe<List<StreamWithState>>
 
     @Query(
         """
@@ -62,7 +63,7 @@ abstract class FeedDAO {
         LIMIT 500
         """
     )
-    abstract fun getAllStreamsForGroup(groupId: Long): Flowable<List<StreamWithState>>
+    abstract fun getAllStreamsForGroup(groupId: Long): Maybe<List<StreamWithState>>
 
     /**
      * @see StreamStateEntity.isFinished()
@@ -97,7 +98,7 @@ abstract class FeedDAO {
         LIMIT 500
         """
     )
-    abstract fun getLiveOrNotPlayedStreams(): Flowable<List<StreamWithState>>
+    abstract fun getLiveOrNotPlayedStreams(): Maybe<List<StreamWithState>>
 
     /**
      * @see StreamStateEntity.isFinished()
@@ -137,7 +138,7 @@ abstract class FeedDAO {
         LIMIT 500
         """
     )
-    abstract fun getLiveOrNotPlayedStreamsForGroup(groupId: Long): Flowable<List<StreamWithState>>
+    abstract fun getLiveOrNotPlayedStreamsForGroup(groupId: Long): Maybe<List<StreamWithState>>
 
     @Query(
         """

app/src/main/java/org/schabi/newpipe/local/feed/FeedDatabaseManager.kt

@@ -42,7 +42,7 @@ class FeedDatabaseManager(context: Context) {
     fun getStreams(
         groupId: Long = FeedGroupEntity.GROUP_ALL_ID,
         getPlayedStreams: Boolean = true
-    ): Flowable<List<StreamWithState>> {
+    ): Maybe<List<StreamWithState>> {
         return when (groupId) {
             FeedGroupEntity.GROUP_ALL_ID -> {
                 if (getPlayedStreams) feedTable.getAllStreams()

app/src/main/java/org/schabi/newpipe/local/feed/FeedViewModel.kt

@@ -33,12 +33,9 @@ class FeedViewModel(
     private var feedDatabaseManager: FeedDatabaseManager = FeedDatabaseManager(applicationContext)
 
     private val toggleShowPlayedItems = BehaviorProcessor.create<Boolean>()
-    private val streamItems = toggleShowPlayedItems
+    private val toggleShowPlayedItemsFlowable = toggleShowPlayedItems
         .startWithItem(initialShowPlayedItems)
         .distinctUntilChanged()
-        .switchMap { showPlayedItems ->
-            feedDatabaseManager.getStreams(groupId, showPlayedItems)
-        }
 
     private val mutableStateLiveData = MutableLiveData<FeedState>()
     val stateLiveData: LiveData<FeedState> = mutableStateLiveData
@@ -46,17 +43,28 @@ class FeedViewModel(
     private var combineDisposable = Flowable
         .combineLatest(
             FeedEventManager.events(),
-            streamItems,
+                toggleShowPlayedItemsFlowable,
             feedDatabaseManager.notLoadedCount(groupId),
             feedDatabaseManager.oldestSubscriptionUpdate(groupId),
 
-            Function4 { t1: FeedEventManager.Event, t2: List<StreamWithState>,
+            Function4 { t1: FeedEventManager.Event, t2: Boolean,
                 t3: Long, t4: List<OffsetDateTime> ->
-                return@Function4 CombineResultHolder(t1, t2, t3, t4.firstOrNull())
+                return@Function4 CombineResultEventHolder(t1, t2, t3, t4.firstOrNull())
             }
         )
         .throttleLatest(DEFAULT_THROTTLE_TIMEOUT, TimeUnit.MILLISECONDS)
         .subscribeOn(Schedulers.io())
+        .observeOn(Schedulers.io())
+        .map { (event, showPlayedItems, notLoadedCount, oldestUpdate) ->
+            var streamItems = if (event is SuccessResultEvent || event is IdleEvent)
+                feedDatabaseManager
+                        .getStreams(groupId, showPlayedItems)
+                        .blockingGet(arrayListOf())
+            else
+                arrayListOf()
+
+            CombineResultDataHolder(event, streamItems, notLoadedCount, oldestUpdate)
+        }
         .observeOn(AndroidSchedulers.mainThread())
         .subscribe { (event, listFromDB, notLoadedCount, oldestUpdate) ->
             mutableStateLiveData.postValue(
@@ -78,7 +86,17 @@ class FeedViewModel(
         combineDisposable.dispose()
     }
 
-    private data class CombineResultHolder(val t1: FeedEventManager.Event, val t2: List<StreamWithState>, val t3: Long, val t4: OffsetDateTime?)
+    private data class CombineResultEventHolder(
+            val t1: FeedEventManager.Event,
+            val t2: Boolean,
+            val t3: Long,
+            val t4: OffsetDateTime?)
+
+    private data class CombineResultDataHolder(
+            val t1: FeedEventManager.Event,
+            val t2: List<StreamWithState>,
+            val t3: Long,
+            val t4: OffsetDateTime?)
 
     fun togglePlayedItems(showPlayedItems: Boolean) {
         toggleShowPlayedItems.onNext(showPlayedItems)