Fix issue with background player

app/src/androidTest/java/org/schabi/newpipe/database/DatabaseMigrationTest.kt

@@ -129,7 +129,7 @@ class DatabaseMigrationTest {
         )
 
         val migratedDatabaseV3 = getMigratedDatabase()
-        val listFromDB = migratedDatabaseV3.streamDAO().all.blockingFirst()
+        val listFromDB = migratedDatabaseV3.streamDAO().getAll().blockingFirst()
 
         // Only expect 2, the one with the null url will be ignored
         assertEquals(2, listFromDB.size)
@@ -217,7 +217,7 @@ class DatabaseMigrationTest {
         )
 
         val migratedDatabaseV8 = getMigratedDatabase()
-        val listFromDB = migratedDatabaseV8.searchHistoryDAO().all.blockingFirst()
+        val listFromDB = migratedDatabaseV8.searchHistoryDAO().getAll().blockingFirst()
 
         assertEquals(2, listFromDB.size)
         assertEquals("abc", listFromDB[0].search)
@@ -283,8 +283,8 @@ class DatabaseMigrationTest {
         )
 
         val migratedDatabaseV9 = getMigratedDatabase()
-        var localListFromDB = migratedDatabaseV9.playlistDAO().all.blockingFirst()
-        var remoteListFromDB = migratedDatabaseV9.playlistRemoteDAO().all.blockingFirst()
+        var localListFromDB = migratedDatabaseV9.playlistDAO().getAll().blockingFirst()
+        var remoteListFromDB = migratedDatabaseV9.playlistRemoteDAO().getAll().blockingFirst()
 
         assertEquals(1, localListFromDB.size)
         assertEquals(localUid2, localListFromDB[0].uid)
@@ -303,8 +303,8 @@ class DatabaseMigrationTest {
             )
         )
 
-        localListFromDB = migratedDatabaseV9.playlistDAO().all.blockingFirst()
-        remoteListFromDB = migratedDatabaseV9.playlistRemoteDAO().all.blockingFirst()
+        localListFromDB = migratedDatabaseV9.playlistDAO().getAll().blockingFirst()
+        remoteListFromDB = migratedDatabaseV9.playlistRemoteDAO().getAll().blockingFirst()
         assertEquals(2, localListFromDB.size)
         assertEquals(localUid3, localListFromDB[1].uid)
         assertEquals(-1, localListFromDB[1].displayIndex)

app/src/androidTest/java/org/schabi/newpipe/local/history/HistoryRecordManagerTest.kt

@@ -41,7 +41,7 @@ class HistoryRecordManagerTest {
         // For some reason the Flowable returned by getAll() never completes, so we can't assert
         // that the number of Lists it returns is exactly 1, we can only check if the first List is
         // correct. Why on earth has a Flowable been used instead of a Single for getAll()?!?
-        val entities = database.searchHistoryDAO().all.blockingFirst()
+        val entities = database.searchHistoryDAO().getAll().blockingFirst()
         assertThat(entities).hasSize(1)
         assertThat(entities[0].id).isEqualTo(1)
         assertThat(entities[0].serviceId).isEqualTo(0)
@@ -59,25 +59,25 @@ class HistoryRecordManagerTest {
 
         // make sure all 4 were inserted
         database.searchHistoryDAO().insertAll(entries)
-        assertThat(database.searchHistoryDAO().all.blockingFirst()).hasSameSizeAs(entries)
+        assertThat(database.searchHistoryDAO().getAll().blockingFirst()).hasSameSizeAs(entries)
 
         // try to delete only "A" entries, "B" entries should be untouched
         manager.deleteSearchHistory("A").test().await().assertValue(2)
-        val entities = database.searchHistoryDAO().all.blockingFirst()
+        val entities = database.searchHistoryDAO().getAll().blockingFirst()
         assertThat(entities).hasSize(2)
         assertThat(entities).usingElementComparator { o1, o2 -> if (o1.hasEqualValues(o2)) 0 else 1 }
             .containsExactly(*entries.subList(2, 4).toTypedArray())
 
         // assert that nothing happens if we delete a search query that does exist in the db
         manager.deleteSearchHistory("A").test().await().assertValue(0)
-        val entities2 = database.searchHistoryDAO().all.blockingFirst()
+        val entities2 = database.searchHistoryDAO().getAll().blockingFirst()
         assertThat(entities2).hasSize(2)
         assertThat(entities2).usingElementComparator { o1, o2 -> if (o1.hasEqualValues(o2)) 0 else 1 }
             .containsExactly(*entries.subList(2, 4).toTypedArray())
 
         // delete all remaining entries
         manager.deleteSearchHistory("B").test().await().assertValue(2)
-        assertThat(database.searchHistoryDAO().all.blockingFirst()).isEmpty()
+        assertThat(database.searchHistoryDAO().getAll().blockingFirst()).isEmpty()
     }
 
     @Test
@@ -90,11 +90,11 @@ class HistoryRecordManagerTest {
 
         // make sure all 3 were inserted
         database.searchHistoryDAO().insertAll(entries)
-        assertThat(database.searchHistoryDAO().all.blockingFirst()).hasSameSizeAs(entries)
+        assertThat(database.searchHistoryDAO().getAll().blockingFirst()).hasSameSizeAs(entries)
 
         // should remove everything
         manager.deleteCompleteSearchHistory().test().await().assertValue(entries.size)
-        assertThat(database.searchHistoryDAO().all.blockingFirst()).isEmpty()
+        assertThat(database.searchHistoryDAO().getAll().blockingFirst()).isEmpty()
     }
 
     private fun insertShuffledRelatedSearches(relatedSearches: Collection<SearchHistoryEntry>) {
@@ -107,7 +107,7 @@ class HistoryRecordManagerTest {
         // make sure all entries were inserted
         assertEquals(
             relatedSearches.size,
-            database.searchHistoryDAO().all.blockingFirst().size
+            database.searchHistoryDAO().getAll().blockingFirst().size
         )
     }
 

app/src/androidTest/java/org/schabi/newpipe/local/playlist/LocalPlaylistManagerTest.kt

@@ -72,6 +72,6 @@ class LocalPlaylistManagerTest {
         val result = manager.createPlaylist("name", listOf(stream, upserted))
 
         result.test().await().assertComplete()
-        database.streamDAO().all.test().awaitCount(1).assertValue(listOf(stream, upserted))
+        database.streamDAO().getAll().test().awaitCount(1).assertValue(listOf(stream, upserted))
     }
 }

app/src/main/java/org/schabi/newpipe/database/BasicDAO.kt

@@ -28,5 +28,5 @@ interface BasicDAO<Entity> {
 
     /* Updates */
     @Update
-    suspend fun update(entity: Entity): Int
+    fun update(entity: Entity): Int
 }

app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamDAO.kt

@@ -34,10 +34,7 @@ interface StreamDAO : BasicDAO<StreamEntity> {
     fun setUploaderUrl(serviceId: Long, url: String, uploaderUrl: String): Completable
 
     @Insert(onConflict = OnConflictStrategy.IGNORE)
-    suspend fun silentInsertInternal(stream: StreamEntity): Long
-
-    @Insert(onConflict = OnConflictStrategy.IGNORE)
-    suspend fun silentInsertAllInternal(streams: List<StreamEntity>): List<Long>
+    fun silentInsertInternal(stream: StreamEntity): Long
 
     @Query("SELECT COUNT(*) != 0 FROM streams WHERE url = :url AND service_id = :serviceId")
     suspend fun exists(serviceId: Int, url: String): Boolean
@@ -48,10 +45,10 @@ interface StreamDAO : BasicDAO<StreamEntity> {
         FROM streams WHERE url = :url AND service_id = :serviceId
         """
     )
-    suspend fun getMinimalStreamForCompare(serviceId: Int, url: String): StreamCompareFeed?
+    fun getMinimalStreamForCompare(serviceId: Int, url: String): StreamCompareFeed?
 
     @Transaction
-    suspend fun upsert(newerStream: StreamEntity): Long {
+    fun upsert(newerStream: StreamEntity): Long {
         val uid = silentInsertInternal(newerStream)
 
         if (uid != -1L) {
@@ -65,20 +62,12 @@ interface StreamDAO : BasicDAO<StreamEntity> {
         return newerStream.uid
     }
 
-    fun upsertBlocking(newerStream: StreamEntity) = runBlocking {
-        upsert(newerStream)
-    }
-
     @Transaction
-    suspend fun upsertAll(streams: List<StreamEntity>): List<Long> {
+    fun upsertAll(streams: List<StreamEntity>): List<Long> {
         return streams.map { upsert(it) }
     }
 
-    fun upsertAllBlocking(streams: List<StreamEntity>) = runBlocking {
-        upsertAll(streams)
-    }
-
-    private suspend fun compareAndUpdateStream(newerStream: StreamEntity) {
+    private fun compareAndUpdateStream(newerStream: StreamEntity) {
         val existentMinimalStream = getMinimalStreamForCompare(newerStream.serviceId, newerStream.url)
             ?: throw IllegalStateException("Stream cannot be null just after insertion.")
         newerStream.uid = existentMinimalStream.uid

app/src/main/java/org/schabi/newpipe/local/history/HistoryRecordManager.java

@@ -110,10 +110,10 @@ public Maybe<Long> markAsWatched(final StreamInfoItem info) {
                         .subscribeOn(Schedulers.io())
                         .blockingGet();
                 duration = completeInfo.getDuration();
-                streamId = streamTable.upsertBlocking(new StreamEntity(completeInfo));
+                streamId = streamTable.upsert(new StreamEntity(completeInfo));
             } else {
                 duration = info.getDuration();
-                streamId = streamTable.upsertBlocking(new StreamEntity(info));
+                streamId = streamTable.upsert(new StreamEntity(info));
             }
 
             // Update the stream progress to the full duration of the video
@@ -141,7 +141,7 @@ public Maybe<Long> onViewed(final StreamInfo info) {
 
         final OffsetDateTime currentTime = OffsetDateTime.now(ZoneOffset.UTC);
         return Maybe.fromCallable(() -> database.runInTransaction(() -> {
-            final long streamId = streamTable.upsertBlocking(new StreamEntity(info));
+            final long streamId = streamTable.upsert(new StreamEntity(info));
             final StreamHistoryEntity latestEntry = streamHistoryTable.getLatestEntry(streamId);
 
             if (latestEntry != null) {
@@ -236,7 +236,7 @@ private boolean isSearchHistoryEnabled() {
 
     public Maybe<StreamStateEntity> loadStreamState(final PlayQueueItem queueItem) {
         return queueItem.getStream()
-                .map(info -> streamTable.upsertBlocking(new StreamEntity(info)))
+                .map(info -> streamTable.upsert(new StreamEntity(info)))
                 .flatMapPublisher(streamStateTable::getState)
                 .firstElement()
                 .flatMap(list -> list.isEmpty() ? Maybe.empty() : Maybe.just(list.get(0)))
@@ -245,7 +245,7 @@ public Maybe<StreamStateEntity> loadStreamState(final PlayQueueItem queueItem) {
     }
 
     public Maybe<StreamStateEntity> loadStreamState(final StreamInfo info) {
-        return Single.fromCallable(() -> streamTable.upsertBlocking(new StreamEntity(info)))
+        return Single.fromCallable(() -> streamTable.upsert(new StreamEntity(info)))
                 .flatMapPublisher(streamStateTable::getState)
                 .firstElement()
                 .flatMap(list -> list.isEmpty() ? Maybe.empty() : Maybe.just(list.get(0)))
@@ -255,7 +255,7 @@ public Maybe<StreamStateEntity> loadStreamState(final StreamInfo info) {
 
     public Completable saveStreamState(@NonNull final StreamInfo info, final long progressMillis) {
         return Completable.fromAction(() -> database.runInTransaction(() -> {
-            final long streamId = streamTable.upsertBlocking(new StreamEntity(info));
+            final long streamId = streamTable.upsert(new StreamEntity(info));
             final StreamStateEntity state = new StreamStateEntity(streamId, progressMillis);
             if (state.isValid(info.getDuration())) {
                 streamStateTable.upsert(state);

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

@@ -46,7 +46,7 @@ public Maybe<List<Long>> createPlaylist(final String name, final List<StreamEnti
         // Make sure the new playlist is always on the top of bookmark.
         // The index will be reassigned to non-negative number in BookmarkFragment.
         return Maybe.fromCallable(() -> database.runInTransaction(() -> {
-                    final List<Long> streamIds = streamTable.upsertAllBlocking(streams);
+                    final List<Long> streamIds = streamTable.upsertAll(streams);
                     final PlaylistEntity newPlaylist = new PlaylistEntity(name, false,
                             streamIds.get(0), -1);
 
@@ -61,7 +61,7 @@ public Maybe<List<Long>> appendToPlaylist(final long playlistId,
         return playlistStreamTable.getMaximumIndexOf(playlistId)
                 .firstElement()
                 .map(maxJoinIndex -> database.runInTransaction(() -> {
-                            final List<Long> streamIds = streamTable.upsertAllBlocking(streams);
+                            final List<Long> streamIds = streamTable.upsertAll(streams);
                             return insertJoinEntities(playlistId, streamIds, maxJoinIndex + 1);
                         }
                 )).subscribeOn(Schedulers.io());

app/src/main/java/org/schabi/newpipe/local/subscription/SubscriptionManager.kt

@@ -81,7 +81,7 @@ class SubscriptionManager(context: Context) {
                         info.description,
                         info.subscriberCount
                     )
-                    runBlocking { subscriptionTable.update(it) }
+                    subscriptionTable.update(it)
                 }
             }
 
@@ -90,7 +90,7 @@ class SubscriptionManager(context: Context) {
             .flatMapCompletable { entity: SubscriptionEntity ->
                 Completable.fromAction {
                     entity.notificationMode = mode
-                    runBlocking { subscriptionTable().update(entity) }
+                    subscriptionTable().update(entity)
                 }.apply {
                     if (mode != NotificationMode.DISABLED) {
                         // notifications have just been enabled, mark all streams as "old"

app/src/main/java/org/schabi/newpipe/util/SparseItemUtil.java

@@ -108,7 +108,7 @@ public static void fetchStreamInfoAndSaveToDatabase(@NonNull final Context conte
                 .subscribe(result -> {
                     // save to database in the background (not on main thread)
                     Completable.fromAction(() -> NewPipeDatabase.getInstance(context)
-                            .streamDAO().upsertBlocking(new StreamEntity(result)))
+                            .streamDAO().upsert(new StreamEntity(result)))
                             .subscribeOn(Schedulers.io())
                             .observeOn(Schedulers.io())
                             .doOnError(throwable ->