Improve DAO method calls

Isira-Seneviratne · Isira-Seneviratne · commit fe563f3307c4 · 2025-03-26T07:57:20.000+05:30
diff --git a/app/src/androidTest/java/org/schabi/newpipe/database/DatabaseMigrationTest.kt b/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)
diff --git a/app/src/androidTest/java/org/schabi/newpipe/database/FeedDAOTest.kt b/app/src/androidTest/java/org/schabi/newpipe/database/FeedDAOTest.kt
@@ -3,7 +3,9 @@ package org.schabi.newpipe.database
 import android.content.Context
 import androidx.room.Room
 import androidx.test.core.app.ApplicationProvider
-import io.reactivex.rxjava3.core.Single
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.rx3.await
 import org.junit.After
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotNull
@@ -22,7 +24,6 @@ import org.schabi.newpipe.extractor.channel.ChannelInfo
 import org.schabi.newpipe.extractor.stream.StreamType
 import java.io.IOException
 import java.time.OffsetDateTime
-import kotlin.streams.toList
 
 class FeedDAOTest {
     private lateinit var db: AppDatabase
@@ -94,14 +95,10 @@ class FeedDAOTest {
         )
     }
 
-    private fun setupUnlinkDelete(time: String) {
+    private fun setupUnlinkDelete(time: String) = runBlocking(Dispatchers.IO) {
         clearAndFillTables()
-        Single.fromCallable {
-            feedDAO.unlinkStreamsOlderThan(OffsetDateTime.parse(time))
-        }.blockingSubscribe()
-        Single.fromCallable {
-            streamDAO.deleteOrphans()
-        }.blockingSubscribe()
+        feedDAO.unlinkStreamsOlderThan(OffsetDateTime.parse(time))
+        streamDAO.deleteOrphans().await()
     }
 
     private fun clearAndFillTables() {
diff --git a/app/src/androidTest/java/org/schabi/newpipe/local/playlist/LocalPlaylistManagerTest.kt b/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))
     }
 }
diff --git a/app/src/main/java/org/schabi/newpipe/database/feed/dao/FeedDAO.kt b/app/src/main/java/org/schabi/newpipe/database/feed/dao/FeedDAO.kt
@@ -119,7 +119,7 @@ abstract class FeedDAO {
               AND   s.upload_date <> max_upload_date))
         """
     )
-    abstract fun unlinkStreamsOlderThan(offsetDateTime: OffsetDateTime)
+    abstract suspend fun unlinkStreamsOlderThan(offsetDateTime: OffsetDateTime)
 
     @Query(
         """
diff --git a/app/src/main/java/org/schabi/newpipe/database/history/dao/StreamHistoryDAO.java b/app/src/main/java/org/schabi/newpipe/database/history/dao/StreamHistoryDAO.java
@@ -30,29 +30,29 @@
 import io.reactivex.rxjava3.core.Flowable;
 
 @Dao
-public abstract class StreamHistoryDAO {
+public interface StreamHistoryDAO {
     @Insert
-    public abstract long insert(StreamHistoryEntity entity);
+    long insert(StreamHistoryEntity entity);
 
     @Delete
-    public abstract void delete(StreamHistoryEntity entity);
+    void delete(StreamHistoryEntity entity);
 
     @Query("DELETE FROM " + STREAM_HISTORY_TABLE)
-    public abstract Completable deleteAll();
+    Completable deleteAll();
 
     @Query("SELECT * FROM " + STREAM_TABLE
             + " INNER JOIN " + STREAM_HISTORY_TABLE
             + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID
             + " ORDER BY " + STREAM_ID + " ASC")
-    public abstract Flowable<List<StreamHistoryEntry>> getHistorySortedById();
+    Flowable<List<StreamHistoryEntry>> getHistorySortedById();
 
     @Query("SELECT * FROM " + STREAM_HISTORY_TABLE + " WHERE " + JOIN_STREAM_ID
             + " = :streamId ORDER BY " + STREAM_ACCESS_DATE + " DESC LIMIT 1")
     @Nullable
-    public abstract StreamHistoryEntity getLatestEntry(long streamId);
+    StreamHistoryEntity getLatestEntry(long streamId);
 
     @Query("DELETE FROM " + STREAM_HISTORY_TABLE + " WHERE " + JOIN_STREAM_ID + " = :streamId")
-    public abstract int deleteStreamHistory(long streamId);
+    Completable deleteStreamHistory(long streamId);
 
     @RewriteQueriesToDropUnusedColumns
     @Query("SELECT * FROM " + STREAM_TABLE
@@ -74,7 +74,7 @@ public abstract class StreamHistoryDAO {
 
             + " ORDER BY " + STREAM_LATEST_DATE + " DESC"
     )
-    public abstract PagingSource<Integer, StreamStatisticsEntry> getHistoryOrderedByLastWatched();
+    PagingSource<Integer, StreamStatisticsEntry> getHistoryOrderedByLastWatched();
 
     @RewriteQueriesToDropUnusedColumns
     @Query("SELECT * FROM " + STREAM_TABLE
@@ -96,5 +96,5 @@ public abstract class StreamHistoryDAO {
 
             + " ORDER BY " + STREAM_WATCH_COUNT + " DESC"
     )
-    public abstract PagingSource<Integer, StreamStatisticsEntry> getHistoryOrderedByViewCount();
+    PagingSource<Integer, StreamStatisticsEntry> getHistoryOrderedByViewCount();
 }
diff --git a/app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamDAO.kt b/app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamDAO.kt
@@ -30,9 +30,6 @@ abstract class StreamDAO : BasicDAO<StreamEntity> {
     @Query("SELECT * FROM streams WHERE url = :url AND service_id = :serviceId")
     abstract fun getStream(serviceId: Long, url: String): Maybe<StreamEntity>
 
-    @Query("UPDATE streams SET uploader_url = :uploaderUrl WHERE url = :url AND service_id = :serviceId")
-    abstract fun setUploaderUrl(serviceId: Long, url: String, uploaderUrl: String): Completable
-
     @Insert(onConflict = OnConflictStrategy.IGNORE)
     internal abstract fun silentInsertInternal(stream: StreamEntity): Long
 
@@ -123,7 +120,7 @@ abstract class StreamDAO : BasicDAO<StreamEntity> {
         WHERE f.stream_id = streams.uid)
         """
     )
-    abstract fun deleteOrphans(): Int
+    abstract fun deleteOrphans(): Completable
 
     /**
      * Minimal entry class used when comparing/updating an existent stream.
diff --git a/app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamStateDAO.kt b/app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamStateDAO.kt
@@ -1,49 +1,23 @@
-package org.schabi.newpipe.database.stream.dao;
+package org.schabi.newpipe.database.stream.dao
 
-import static org.schabi.newpipe.database.stream.model.StreamStateEntity.JOIN_STREAM_ID;
-import static org.schabi.newpipe.database.stream.model.StreamStateEntity.STREAM_STATE_TABLE;
-
-import androidx.room.Dao;
-import androidx.room.Insert;
-import androidx.room.OnConflictStrategy;
-import androidx.room.Query;
-import androidx.room.Transaction;
-
-import org.schabi.newpipe.database.BasicDAO;
-import org.schabi.newpipe.database.stream.model.StreamStateEntity;
-
-import java.util.List;
-
-import io.reactivex.rxjava3.core.Flowable;
-import io.reactivex.rxjava3.core.Maybe;
+import androidx.room.Dao
+import androidx.room.Query
+import androidx.room.Upsert
+import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Maybe
+import org.schabi.newpipe.database.stream.model.StreamStateEntity
 
 @Dao
-public interface StreamStateDAO extends BasicDAO<StreamStateEntity> {
-    @Override
-    @Query("SELECT * FROM " + STREAM_STATE_TABLE)
-    Flowable<List<StreamStateEntity>> getAll();
-
-    @Override
-    @Query("DELETE FROM " + STREAM_STATE_TABLE)
-    int deleteAll();
-
-    @Override
-    default Flowable<List<StreamStateEntity>> listByService(final int serviceId) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Query("SELECT * FROM " + STREAM_STATE_TABLE + " WHERE " + JOIN_STREAM_ID + " = :streamId")
-    Maybe<StreamStateEntity> getState(long streamId);
+interface StreamStateDAO {
+    @Query("DELETE FROM " + StreamStateEntity.STREAM_STATE_TABLE)
+    fun deleteAll(): Completable
 
-    @Query("DELETE FROM " + STREAM_STATE_TABLE + " WHERE " + JOIN_STREAM_ID + " = :streamId")
-    int deleteState(long streamId);
+    @Query("SELECT * FROM " + StreamStateEntity.STREAM_STATE_TABLE + " WHERE " + StreamStateEntity.JOIN_STREAM_ID + " = :streamId")
+    fun getState(streamId: Long): Maybe<StreamStateEntity>
 
-    @Insert(onConflict = OnConflictStrategy.IGNORE)
-    void silentInsertInternal(StreamStateEntity streamState);
+    @Query("DELETE FROM " + StreamStateEntity.STREAM_STATE_TABLE + " WHERE " + StreamStateEntity.JOIN_STREAM_ID + " = :streamId")
+    fun deleteState(streamId: Long): Completable
 
-    @Transaction
-    default long upsert(final StreamStateEntity stream) {
-        silentInsertInternal(stream);
-        return update(stream);
-    }
+    @Upsert
+    fun upsert(stream: StreamStateEntity)
 }
diff --git a/app/src/main/java/org/schabi/newpipe/local/feed/FeedDatabaseManager.kt b/app/src/main/java/org/schabi/newpipe/local/feed/FeedDatabaseManager.kt
@@ -1,13 +1,12 @@
 package org.schabi.newpipe.local.feed
 
 import android.content.Context
-import android.util.Log
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Maybe
 import io.reactivex.rxjava3.schedulers.Schedulers
-import org.schabi.newpipe.MainActivity.DEBUG
+import kotlinx.coroutines.rx3.await
 import org.schabi.newpipe.NewPipeDatabase
 import org.schabi.newpipe.database.feed.model.FeedEntity
 import org.schabi.newpipe.database.feed.model.FeedGroupEntity
@@ -111,20 +110,9 @@ class FeedDatabaseManager(context: Context) {
         )
     }
 
-    fun removeOrphansOrOlderStreams(oldestAllowedDate: OffsetDateTime = FEED_OLDEST_ALLOWED_DATE) {
+    suspend fun removeOrphansOrOlderStreams(oldestAllowedDate: OffsetDateTime = FEED_OLDEST_ALLOWED_DATE) {
         feedTable.unlinkStreamsOlderThan(oldestAllowedDate)
-        streamTable.deleteOrphans()
-    }
-
-    fun clear() {
-        feedTable.deleteAll()
-        val deletedOrphans = streamTable.deleteOrphans()
-        if (DEBUG) {
-            Log.d(
-                this::class.java.simpleName,
-                "clear() → streamTable.deleteOrphans() → $deletedOrphans"
-            )
-        }
+        streamTable.deleteOrphans().await()
     }
 
     // /////////////////////////////////////////////////////////////////////////
diff --git a/app/src/main/java/org/schabi/newpipe/local/feed/service/FeedLoadManager.kt b/app/src/main/java/org/schabi/newpipe/local/feed/service/FeedLoadManager.kt
@@ -4,13 +4,14 @@ import android.content.Context
 import android.content.SharedPreferences
 import androidx.preference.PreferenceManager
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
-import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Notification
 import io.reactivex.rxjava3.core.Single
 import io.reactivex.rxjava3.functions.Consumer
 import io.reactivex.rxjava3.processors.PublishProcessor
 import io.reactivex.rxjava3.schedulers.Schedulers
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.rx3.rxCompletable
 import org.schabi.newpipe.R
 import org.schabi.newpipe.database.feed.model.FeedGroupEntity
 import org.schabi.newpipe.database.subscription.NotificationMode
@@ -256,7 +257,7 @@ class FeedLoadManager(private val context: Context) {
      * Remove streams from the feed which are older than [FeedDatabaseManager.FEED_OLDEST_ALLOWED_DATE].
      * Remove streams from the database which are not linked / used by any table.
      */
-    private fun postProcessFeed() = Completable.fromRunnable {
+    private fun postProcessFeed() = rxCompletable(Dispatchers.IO) {
         FeedEventManager.postEvent(FeedEventManager.Event.ProgressEvent(R.string.feed_processing_message))
         feedDatabaseManager.removeOrphansOrOlderStreams()
 
diff --git a/app/src/main/java/org/schabi/newpipe/local/history/HistoryRecordManager.java b/app/src/main/java/org/schabi/newpipe/local/history/HistoryRecordManager.java
@@ -151,19 +151,16 @@ public Maybe<Long> onViewed(final StreamInfo info) {
     }
 
     public Completable deleteStreamHistoryAndState(final long streamId) {
-        return Completable.fromAction(() -> {
-            streamStateTable.deleteState(streamId);
-            streamHistoryTable.deleteStreamHistory(streamId);
-        }).subscribeOn(Schedulers.io());
+        return streamStateTable.deleteState(streamId)
+                .andThen(streamHistoryTable.deleteStreamHistory(streamId));
     }
 
     public Completable deleteWholeStreamHistory() {
         return streamHistoryTable.deleteAll().subscribeOn(Schedulers.io());
     }
 
-    public Single<Integer> deleteCompleteStreamStateHistory() {
-        return Single.fromCallable(streamStateTable::deleteAll)
-                .subscribeOn(Schedulers.io());
+    public Completable deleteCompleteStreamStateHistory() {
+        return streamStateTable.deleteAll().subscribeOn(Schedulers.io());
     }
 
     public Flowable<List<StreamHistoryEntry>> getStreamHistorySortedById() {
@@ -278,8 +275,7 @@ public Single<List<StreamStateEntity>> loadLocalStreamStateBatch(
     // Utility
     ///////////////////////////////////////////////////////
 
-    public Single<Integer> removeOrphanedRecords() {
-        return Single.fromCallable(streamTable::deleteOrphans).subscribeOn(Schedulers.io());
+    public Completable removeOrphanedRecords() {
+        return streamTable.deleteOrphans().subscribeOn(Schedulers.io());
     }
-
 }
diff --git a/app/src/main/java/org/schabi/newpipe/settings/HistorySettingsFragment.java b/app/src/main/java/org/schabi/newpipe/settings/HistorySettingsFragment.java
@@ -20,6 +20,7 @@
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
 import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.internal.functions.Functions;
 
 public class HistorySettingsFragment extends BasePreferenceFragment {
     private String cacheWipeKey;
@@ -79,8 +80,8 @@ private static Disposable getDeletePlaybackStatesDisposable(
         return recordManager.deleteCompleteStreamStateHistory()
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(
-                        howManyDeleted -> Toast.makeText(context,
-                                R.string.watch_history_states_deleted,  Toast.LENGTH_SHORT).show(),
+                        () -> Toast.makeText(context, R.string.watch_history_states_deleted,
+                                Toast.LENGTH_SHORT).show(),
                         throwable -> ErrorUtil.openActivity(context,
                                 new ErrorInfo(throwable, UserAction.DELETE_FROM_HISTORY,
                                         "Delete playback states")));
@@ -103,7 +104,7 @@ private static Disposable getRemoveOrphanedRecordsDisposable(
         return recordManager.removeOrphanedRecords()
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(
-                        howManyDeleted -> { },
+                        Functions.EMPTY_ACTION,
                         throwable -> ErrorUtil.openActivity(context,
                                 new ErrorInfo(throwable, UserAction.DELETE_FROM_HISTORY,
                                         "Clear orphaned records")));
diff --git a/app/src/main/java/org/schabi/newpipe/viewmodels/StreamViewModel.kt b/app/src/main/java/org/schabi/newpipe/viewmodels/StreamViewModel.kt
@@ -3,6 +3,7 @@ package org.schabi.newpipe.viewmodels
 import android.app.Application
 import androidx.lifecycle.AndroidViewModel
 import androidx.lifecycle.viewModelScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.rx3.await
 import kotlinx.coroutines.rx3.awaitSingleOrNull
@@ -25,7 +26,7 @@ class StreamViewModel(application: Application) : AndroidViewModel(application)
     }
 
     fun deleteStreamHistory(streamId: Long) {
-        viewModelScope.launch {
+        viewModelScope.launch(Dispatchers.IO) {
             historyRecordManager.deleteStreamHistoryAndState(streamId).await()
         }
     }

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,7 @@ class DatabaseMigrationTest {`
`129`	`129`	`)`
`130`	`130`
`131`	`131`	`val migratedDatabaseV3 = getMigratedDatabase()`
`132`		`- val listFromDB = migratedDatabaseV3.streamDAO().all.blockingFirst()`
	`132`	`+ val listFromDB = migratedDatabaseV3.streamDAO().getAll().blockingFirst()`
`133`	`133`
`134`	`134`	`// Only expect 2, the one with the null url will be ignored`
`135`	`135`	`assertEquals(2, listFromDB.size)`
Original file line number	Diff line number	Diff line change
`@@ -72,6 +72,6 @@ class LocalPlaylistManagerTest {`
`72`	`72`	`val result = manager.createPlaylist("name", listOf(stream, upserted))`
`73`	`73`
`74`	`74`	`result.test().await().assertComplete()`
`75`		`- database.streamDAO().all.test().awaitCount(1).assertValue(listOf(stream, upserted))`
	`75`	`+ database.streamDAO().getAll().test().awaitCount(1).assertValue(listOf(stream, upserted))`
`76`	`76`	`}`
`77`	`77`	`}`
Original file line number	Diff line number	Diff line change
`@@ -119,7 +119,7 @@ abstract class FeedDAO {`
`119`	`119`	`AND s.upload_date <> max_upload_date))`
`120`	`120`	`"""`
`121`	`121`	`)`
`122`		`- abstract fun unlinkStreamsOlderThan(offsetDateTime: OffsetDateTime)`
	`122`	`+ abstract suspend fun unlinkStreamsOlderThan(offsetDateTime: OffsetDateTime)`
`123`	`123`
`124`	`124`	`@Query(`
`125`	`125`	`"""`
Original file line number	Diff line number	Diff line change
`@@ -151,19 +151,16 @@ public Maybe<Long> onViewed(final StreamInfo info) {`
`151`	`151`	`}`
`152`	`152`
`153`	`153`	`public Completable deleteStreamHistoryAndState(final long streamId) {`
`154`		`- return Completable.fromAction(() -> {`
`155`		`- streamStateTable.deleteState(streamId);`
`156`		`- streamHistoryTable.deleteStreamHistory(streamId);`
`157`		`- }).subscribeOn(Schedulers.io());`
	`154`	`+ return streamStateTable.deleteState(streamId)`
	`155`	`+ .andThen(streamHistoryTable.deleteStreamHistory(streamId));`
`158`	`156`	`}`
`159`	`157`
`160`	`158`	`public Completable deleteWholeStreamHistory() {`
`161`	`159`	`return streamHistoryTable.deleteAll().subscribeOn(Schedulers.io());`
`162`	`160`	`}`
`163`	`161`
`164`		`- public Single<Integer> deleteCompleteStreamStateHistory() {`
`165`		`- return Single.fromCallable(streamStateTable::deleteAll)`
`166`		`- .subscribeOn(Schedulers.io());`
	`162`	`+ public Completable deleteCompleteStreamStateHistory() {`
	`163`	`+ return streamStateTable.deleteAll().subscribeOn(Schedulers.io());`
`167`	`164`	`}`
`168`	`165`
`169`	`166`	`public Flowable<List<StreamHistoryEntry>> getStreamHistorySortedById() {`
`@@ -278,8 +275,7 @@ public Single<List<StreamStateEntity>> loadLocalStreamStateBatch(`
`278`	`275`	`// Utility`
`279`	`276`	`///////////////////////////////////////////////////////`
`280`	`277`
`281`		`- public Single<Integer> removeOrphanedRecords() {`
`282`		`- return Single.fromCallable(streamTable::deleteOrphans).subscribeOn(Schedulers.io());`
	`278`	`+ public Completable removeOrphanedRecords() {`
	`279`	`+ return streamTable.deleteOrphans().subscribeOn(Schedulers.io());`
`283`	`280`	`}`
`284`		`-`
`285`	`281`	`}`