Merge pull request #386 from chrisbanes/cb/room-flow

Use Room's new Flow support

build.gradle

@@ -61,6 +61,10 @@ allprojects {
         google()
         mavenCentral()
         jcenter()
+
+        // AndroidX snapshots. Remove this once Room 2.2.0-alpha03 is released
+        maven { url 'https://ci.android.com/builds/submitted/5799541/androidx_snapshot/latest/repository' }
+
         mavenLocal()
     }
 }

buildSrc/src/main/java/app/tivi/buildsrc/dependencies.kt

@@ -122,10 +122,9 @@ object Libs {
         }
 
         object Room {
-            private const val version = "2.1.0"
+            private const val version = "2.2.0-SNAPSHOT"
             const val common = "androidx.room:room-common:$version"
             const val runtime = "androidx.room:room-runtime:$version"
-            const val rxjava2 = "androidx.room:room-rxjava2:$version"
             const val compiler = "androidx.room:room-compiler:$version"
             const val ktx = "androidx.room:room-ktx:$version"
             const val testing = "androidx.room:room-testing:$version"

data-android/build.gradle

@@ -68,7 +68,6 @@ dependencies {
     implementation project(":data")
 
     implementation Libs.AndroidX.Room.runtime
-    implementation Libs.AndroidX.Room.rxjava2
     kapt Libs.AndroidX.Room.compiler
 
     implementation Libs.AndroidX.Paging.runtime

data/build.gradle

@@ -35,10 +35,6 @@ dependencies {
 
     api Libs.AndroidX.Paging.common
 
-    implementation Libs.Coroutines.rx2
-
-    implementation Libs.RxJava.rxJava
-
     api Libs.threeTenBpNoTzdb
 
     kapt Libs.Dagger.compiler

data/src/main/java/app/tivi/data/daos/EpisodeWatchEntryDao.kt

@@ -20,7 +20,7 @@ import androidx.room.Dao
 import androidx.room.Query
 import app.tivi.data.entities.EpisodeWatchEntry
 import app.tivi.data.entities.PendingAction
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class EpisodeWatchEntryDao : EntityDao<EpisodeWatchEntry> {
@@ -31,7 +31,7 @@ abstract class EpisodeWatchEntryDao : EntityDao<EpisodeWatchEntry> {
     abstract suspend fun watchCountForEpisode(episodeId: Long): Int
 
     @Query("SELECT * FROM episode_watch_entries WHERE episode_id = :episodeId")
-    abstract fun watchesForEpisodeObservable(episodeId: Long): Flowable<List<EpisodeWatchEntry>>
+    abstract fun watchesForEpisodeObservable(episodeId: Long): Flow<List<EpisodeWatchEntry>>
 
     @Query("SELECT * FROM episode_watch_entries WHERE id = :id")
     abstract suspend fun entryWithId(id: Long): EpisodeWatchEntry?

data/src/main/java/app/tivi/data/daos/EpisodesDao.kt

@@ -19,8 +19,8 @@ package app.tivi.data.daos
 import androidx.room.Dao
 import androidx.room.Query
 import app.tivi.data.entities.Episode
-import io.reactivex.Flowable
 import app.tivi.data.resultentities.EpisodeWithSeason
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class EpisodesDao : EntityDao<Episode> {
@@ -46,7 +46,7 @@ abstract class EpisodesDao : EntityDao<Episode> {
     abstract suspend fun episodeIdWithTraktId(traktId: Int): Long?
 
     @Query("SELECT * from episodes WHERE id = :id")
-    abstract fun episodeWithIdObservable(id: Long): Flowable<Episode>
+    abstract fun episodeWithIdObservable(id: Long): Flow<Episode>
 
     @Query("SELECT shows.id FROM shows" +
             " INNER JOIN seasons AS s ON s.show_id = shows.id" +
@@ -55,13 +55,13 @@ abstract class EpisodesDao : EntityDao<Episode> {
     abstract suspend fun showIdForEpisodeId(episodeId: Long): Long
 
     @Query(latestWatchedEpisodeForShowId)
-    abstract fun latestWatchedEpisodeForShowId(showId: Long): Flowable<EpisodeWithSeason?>
+    abstract fun latestWatchedEpisodeForShowId(showId: Long): Flow<EpisodeWithSeason>
 
     @Query(nextEpisodeForShowIdAfter)
-    abstract fun nextEpisodeForShowAfter(showId: Long, seasonNumber: Int, episodeNumber: Int): Flowable<EpisodeWithSeason>
+    abstract fun nextEpisodeForShowAfter(showId: Long, seasonNumber: Int, episodeNumber: Int): Flow<EpisodeWithSeason>
 
     @Query(nextAiredEpisodeForShowIdAfter)
-    abstract fun nextAiredEpisodeForShowAfter(showId: Long, seasonNumber: Int, episodeNumber: Int): Flowable<EpisodeWithSeason>
+    abstract fun nextAiredEpisodeForShowAfter(showId: Long, seasonNumber: Int, episodeNumber: Int): Flow<EpisodeWithSeason>
 
     companion object {
         const val latestWatchedEpisodeForShowId = """

data/src/main/java/app/tivi/data/daos/FollowedShowsDao.kt

@@ -24,7 +24,7 @@ import app.tivi.data.entities.FollowedShowEntry
 import app.tivi.data.entities.PendingAction
 import app.tivi.data.entities.Season
 import app.tivi.data.resultentities.FollowedShowEntryWithShow
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class FollowedShowsDao : EntryDao<FollowedShowEntry, FollowedShowEntryWithShow> {
@@ -74,7 +74,7 @@ abstract class FollowedShowsDao : EntryDao<FollowedShowEntry, FollowedShowEntryW
     abstract suspend fun entryWithShowId(showId: Long): FollowedShowEntry?
 
     @Query("SELECT COUNT(*) FROM myshows_entries WHERE show_id = :showId AND pending_action != 'delete'")
-    abstract fun entryCountWithShowIdNotPendingDeleteObservable(showId: Long): Flowable<Int>
+    abstract fun entryCountWithShowIdNotPendingDeleteObservable(showId: Long): Flow<Int>
 
     @Query("SELECT COUNT(*) FROM myshows_entries WHERE show_id = :showId")
     abstract suspend fun entryCountWithShowId(showId: Long): Int

data/src/main/java/app/tivi/data/daos/PairEntryDao.kt

@@ -18,14 +18,14 @@ package app.tivi.data.daos
 
 import app.tivi.data.MultipleEntry
 import app.tivi.data.resultentities.EntryWithShow
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 /**
  * This interface represents a DAO which contains entities which are part of a collective list for a given show.
  */
 interface PairEntryDao<EC : MultipleEntry, LI : EntryWithShow<EC>> : EntityDao<EC> {
     fun entries(showId: Long): List<EC>
     fun entriesWithShows(showId: Long): List<LI>
-    fun entriesWithShowsObservable(showId: Long): Flowable<List<LI>>
+    fun entriesWithShowsObservable(showId: Long): Flow<List<LI>>
     suspend fun deleteWithShowId(showId: Long)
 }

data/src/main/java/app/tivi/data/daos/PopularDao.kt

@@ -22,13 +22,13 @@ import androidx.room.Query
 import androidx.room.Transaction
 import app.tivi.data.entities.PopularShowEntry
 import app.tivi.data.resultentities.PopularEntryWithShow
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class PopularDao : PaginatedEntryDao<PopularShowEntry, PopularEntryWithShow> {
     @Transaction
     @Query("SELECT * FROM popular_shows ORDER BY page, page_order LIMIT :count OFFSET :offset")
-    abstract fun entriesObservable(count: Int, offset: Int): Flowable<List<PopularEntryWithShow>>
+    abstract fun entriesObservable(count: Int, offset: Int): Flow<List<PopularEntryWithShow>>
 
     @Transaction
     @Query("SELECT * FROM popular_shows ORDER BY page, page_order")

data/src/main/java/app/tivi/data/daos/RelatedShowsDao.kt

@@ -21,7 +21,7 @@ import androidx.room.Query
 import androidx.room.Transaction
 import app.tivi.data.entities.RelatedShowEntry
 import app.tivi.data.resultentities.RelatedShowEntryWithShow
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class RelatedShowsDao : PairEntryDao<RelatedShowEntry, RelatedShowEntryWithShow> {
@@ -35,7 +35,7 @@ abstract class RelatedShowsDao : PairEntryDao<RelatedShowEntry, RelatedShowEntry
 
     @Transaction
     @Query("SELECT * FROM related_shows WHERE show_id = :showId ORDER BY order_index")
-    abstract override fun entriesWithShowsObservable(showId: Long): Flowable<List<RelatedShowEntryWithShow>>
+    abstract override fun entriesWithShowsObservable(showId: Long): Flow<List<RelatedShowEntryWithShow>>
 
     @Query("DELETE FROM related_shows WHERE show_id = :showId")
     abstract override suspend fun deleteWithShowId(showId: Long)

data/src/main/java/app/tivi/data/daos/SeasonsDao.kt

@@ -22,13 +22,13 @@ import androidx.room.Transaction
 import app.tivi.data.entities.Season
 import app.tivi.data.entities.Season.Companion.NUMBER_SPECIALS
 import app.tivi.data.resultentities.SeasonWithEpisodesAndWatches
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class SeasonsDao : EntityDao<Season> {
     @Transaction
     @Query("SELECT * FROM seasons WHERE show_id = :showId ORDER BY number=$NUMBER_SPECIALS, number")
-    abstract fun seasonsWithEpisodesForShowId(showId: Long): Flowable<List<SeasonWithEpisodesAndWatches>>
+    abstract fun seasonsWithEpisodesForShowId(showId: Long): Flow<List<SeasonWithEpisodesAndWatches>>
 
     @Query("SELECT * FROM seasons WHERE show_id = :showId ORDER BY number=$NUMBER_SPECIALS, number")
     abstract suspend fun seasonsForShowId(showId: Long): List<Season>

data/src/main/java/app/tivi/data/daos/TiviShowDao.kt

@@ -20,21 +20,21 @@ import androidx.room.Dao
 import androidx.room.Query
 import app.tivi.data.entities.TiviShow
 import app.tivi.data.resultentities.ShowDetailed
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class TiviShowDao : EntityDao<TiviShow> {
     @Query("SELECT * FROM shows WHERE trakt_id = :id")
     abstract suspend fun getShowWithTraktId(id: Int): TiviShow?
 
     @Query("SELECT * FROM shows WHERE id IN (:ids)")
-    abstract fun getShowsWithIds(ids: List<Long>): Flowable<List<TiviShow>>
+    abstract fun getShowsWithIds(ids: List<Long>): Flow<List<TiviShow>>
 
     @Query("SELECT * FROM shows WHERE tmdb_id = :id")
     abstract suspend fun getShowWithTmdbId(id: Int): TiviShow?
 
     @Query("SELECT * FROM shows WHERE id = :id")
-    abstract fun getShowWithIdFlowable(id: Long): Flowable<ShowDetailed>
+    abstract fun getShowWithIdFlow(id: Long): Flow<ShowDetailed>
 
     @Query("SELECT * FROM shows WHERE id = :id")
     abstract suspend fun getShowWithId(id: Long): TiviShow?

data/src/main/java/app/tivi/data/daos/TrendingDao.kt

@@ -22,13 +22,13 @@ import androidx.room.Query
 import androidx.room.Transaction
 import app.tivi.data.entities.TrendingShowEntry
 import app.tivi.data.resultentities.TrendingEntryWithShow
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class TrendingDao : PaginatedEntryDao<TrendingShowEntry, TrendingEntryWithShow> {
     @Transaction
     @Query("SELECT * FROM trending_shows ORDER BY page ASC, watchers DESC, id ASC LIMIT :count OFFSET :offset")
-    abstract fun entriesObservable(count: Int, offset: Int): Flowable<List<TrendingEntryWithShow>>
+    abstract fun entriesObservable(count: Int, offset: Int): Flow<List<TrendingEntryWithShow>>
 
     @Transaction
     @Query("SELECT * FROM trending_shows ORDER BY page ASC, watchers DESC, id ASC")

data/src/main/java/app/tivi/data/daos/UserDao.kt

@@ -21,15 +21,15 @@ import androidx.room.Insert
 import androidx.room.OnConflictStrategy
 import androidx.room.Query
 import app.tivi.data.entities.TraktUser
-import io.reactivex.Flowable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 interface UserDao : EntityDao<TraktUser> {
     @Query("SELECT * FROM users WHERE is_me != 0")
-    fun observeMe(): Flowable<TraktUser>
+    fun observeMe(): Flow<TraktUser>
 
     @Query("SELECT * FROM users WHERE username = :username")
-    fun observeTraktUser(username: String): Flowable<TraktUser>
+    fun observeTraktUser(username: String): Flow<TraktUser>
 
     @Query("SELECT * FROM users WHERE username = :username")
     suspend fun getTraktUser(username: String): TraktUser?

data/src/main/java/app/tivi/data/daos/WatchedShowDao.kt

@@ -22,7 +22,7 @@ import androidx.room.Query
 import androidx.room.Transaction
 import app.tivi.data.entities.WatchedShowEntry
 import app.tivi.data.resultentities.WatchedShowEntryWithShow
-import io.reactivex.Observable
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 abstract class WatchedShowDao : EntryDao<WatchedShowEntry, WatchedShowEntryWithShow> {
@@ -36,7 +36,7 @@ abstract class WatchedShowDao : EntryDao<WatchedShowEntry, WatchedShowEntryWithS
 
     @Transaction
     @Query("$ENTRY_QUERY_ORDER_LAST_WATCHED LIMIT :count OFFSET :offset")
-    abstract fun entriesObservable(count: Int, offset: Int): Observable<List<WatchedShowEntryWithShow>>
+    abstract fun entriesObservable(count: Int, offset: Int): Flow<List<WatchedShowEntryWithShow>>
 
     @Transaction
     @Query(ENTRY_QUERY_ORDER_LAST_WATCHED)

data/src/main/java/app/tivi/data/repositories/episodes/EpisodeWatchStore.kt

@@ -24,7 +24,6 @@ import app.tivi.data.entities.PendingAction
 import app.tivi.data.syncers.syncerForEntity
 import app.tivi.util.Logger
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.reactive.asFlow
 import javax.inject.Inject
 
 class EpisodeWatchStore @Inject constructor(
@@ -41,7 +40,7 @@ class EpisodeWatchStore @Inject constructor(
     )
 
     fun observeEpisodeWatches(episodeId: Long): Flow<List<EpisodeWatchEntry>> {
-        return episodeWatchEntryDao.watchesForEpisodeObservable(episodeId).asFlow()
+        return episodeWatchEntryDao.watchesForEpisodeObservable(episodeId)
     }
 
     suspend fun save(watch: EpisodeWatchEntry) = entityInserter.insertOrUpdate(episodeWatchEntryDao, watch)

data/src/main/java/app/tivi/data/repositories/episodes/SeasonsEpisodesStore.kt

@@ -27,7 +27,7 @@ import app.tivi.data.resultentities.SeasonWithEpisodesAndWatches
 import app.tivi.data.syncers.syncerForEntity
 import app.tivi.util.Logger
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.reactive.asFlow
+import kotlinx.coroutines.flow.flatMapConcat
 import javax.inject.Inject
 
 class SeasonsEpisodesStore @Inject constructor(
@@ -52,21 +52,20 @@ class SeasonsEpisodesStore @Inject constructor(
     )
 
     fun observeEpisode(episodeId: Long): Flow<Episode> {
-        return episodesDao.episodeWithIdObservable(episodeId).asFlow()
+        return episodesDao.episodeWithIdObservable(episodeId)
     }
 
     fun observeShowSeasonsWithEpisodes(showId: Long): Flow<List<SeasonWithEpisodesAndWatches>> {
-        return seasonsDao.seasonsWithEpisodesForShowId(showId).asFlow()
+        return seasonsDao.seasonsWithEpisodesForShowId(showId)
     }
 
     fun observeShowNextEpisodeToWatch(showId: Long): Flow<EpisodeWithSeason> {
-        return episodesDao.latestWatchedEpisodeForShowId(showId)
-                .flatMap {
-                    episodesDao.nextAiredEpisodeForShowAfter(showId,
-                            it.season!!.number!!,
-                            it.episode!!.number!!
-                    )
-                }.asFlow()
+        return episodesDao.latestWatchedEpisodeForShowId(showId).flatMapConcat {
+            episodesDao.nextAiredEpisodeForShowAfter(showId,
+                    it.season!!.number!!,
+                    it.episode!!.number!!
+            )
+        }
     }
 
     /**

data/src/main/java/app/tivi/data/repositories/followedshows/FollowedShowsStore.kt

@@ -28,7 +28,6 @@ import app.tivi.data.syncers.syncerForEntity
 import app.tivi.util.Logger
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.reactive.asFlow
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -98,7 +97,6 @@ class FollowedShowsStore @Inject constructor(
 
     fun observeIsShowFollowed(showId: Long): Flow<Boolean> {
         return followedShowsDao.entryCountWithShowIdNotPendingDeleteObservable(showId)
-                .asFlow()
                 .map { it > 0 }
     }
 

data/src/main/java/app/tivi/data/repositories/popularshows/PopularShowsRepository.kt

@@ -23,7 +23,6 @@ import app.tivi.data.repositories.shows.ShowRepository
 import app.tivi.data.resultentities.PopularEntryWithShow
 import app.tivi.extensions.parallelForEach
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.reactive.asFlow
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -37,7 +36,7 @@ class PopularShowsRepository @Inject constructor(
     fun observeForPaging(): DataSource.Factory<Int, PopularEntryWithShow> = popularShowsStore.observeForPaging()
 
     fun observeForObservable(): Flow<List<PopularEntryWithShow>> {
-        return popularShowsStore.observeForObservable(15, 0).asFlow()
+        return popularShowsStore.observeForObservable(15, 0)
     }
 
     suspend fun loadNextPage() {

data/src/main/java/app/tivi/data/repositories/popularshows/PopularShowsStore.kt

@@ -21,16 +21,13 @@ import app.tivi.data.DatabaseTransactionRunner
 import app.tivi.data.daos.PopularDao
 import app.tivi.data.entities.PopularShowEntry
 import app.tivi.data.resultentities.PopularEntryWithShow
-import io.reactivex.Flowable
 import javax.inject.Inject
 
 class PopularShowsStore @Inject constructor(
     private val transactionRunner: DatabaseTransactionRunner,
     private val popularShowDao: PopularDao
 ) {
-    fun observeForObservable(count: Int, offset: Int): Flowable<List<PopularEntryWithShow>> {
-        return popularShowDao.entriesObservable(count, offset)
-    }
+    fun observeForObservable(count: Int, offset: Int) = popularShowDao.entriesObservable(count, offset)
 
     fun observeForPaging(): DataSource.Factory<Int, PopularEntryWithShow> {
         return popularShowDao.entriesDataSource()

data/src/main/java/app/tivi/data/repositories/relatedshows/RelatedShowsStore.kt

@@ -22,7 +22,6 @@ import app.tivi.data.daos.RelatedShowsDao
 import app.tivi.data.entities.RelatedShowEntry
 import app.tivi.data.resultentities.RelatedShowEntryWithShow
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.reactive.asFlow
 import javax.inject.Inject
 
 class RelatedShowsStore @Inject constructor(
@@ -33,7 +32,7 @@ class RelatedShowsStore @Inject constructor(
     suspend fun getRelatedShows(showId: Long) = relatedShowsDao.entries(showId)
 
     fun observeRelatedShows(showId: Long): Flow<List<RelatedShowEntryWithShow>> {
-        return relatedShowsDao.entriesWithShowsObservable(showId).asFlow()
+        return relatedShowsDao.entriesWithShowsObservable(showId)
     }
 
     suspend fun saveRelatedShows(showId: Long, relatedShows: List<RelatedShowEntry>) = transactionRunner {