Fix notification profile UI state bug and crash.

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/notifications/profiles/NotificationProfilesRepository.kt

@@ -1,6 +1,7 @@
 package org.thoughtcrime.securesms.components.settings.app.notifications.profiles
 
 import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Observable
 import io.reactivex.rxjava3.core.ObservableEmitter
 import io.reactivex.rxjava3.core.Single
@@ -8,6 +9,7 @@ import io.reactivex.rxjava3.schedulers.Schedulers
 import org.thoughtcrime.securesms.conversation.colors.AvatarColor
 import org.thoughtcrime.securesms.database.DatabaseObserver
 import org.thoughtcrime.securesms.database.NotificationProfileDatabase
+import org.thoughtcrime.securesms.database.RxDatabaseObserver
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.keyvalue.SignalStore
@@ -24,16 +26,11 @@ import org.thoughtcrime.securesms.util.toMillis
 class NotificationProfilesRepository {
   private val database: NotificationProfileDatabase = SignalDatabase.notificationProfiles
 
-  fun getProfiles(): Observable<List<NotificationProfile>> {
-    return Observable.create { emitter: ObservableEmitter<List<NotificationProfile>> ->
-      val databaseObserver: DatabaseObserver = ApplicationDependencies.getDatabaseObserver()
-      val profileObserver = DatabaseObserver.Observer { emitter.onNext(database.getProfiles()) }
-
-      databaseObserver.registerNotificationProfileObserver(profileObserver)
-
-      emitter.setCancellable { databaseObserver.unregisterObserver(profileObserver) }
-      emitter.onNext(database.getProfiles())
-    }.subscribeOn(Schedulers.io())
+  fun getProfiles(): Flowable<List<NotificationProfile>> {
+    return RxDatabaseObserver
+      .notificationProfiles
+      .map { database.getProfiles() }
+      .subscribeOn(Schedulers.io())
   }
 
   fun getProfile(profileId: Long): Observable<NotificationProfile> {

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/notifications/profiles/NotificationProfilesViewModel.kt

@@ -3,12 +3,12 @@ package org.thoughtcrime.securesms.components.settings.app.notifications.profile
 import androidx.lifecycle.ViewModel
 import androidx.lifecycle.ViewModelProvider
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
-import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.core.Flowable
 import org.thoughtcrime.securesms.notifications.profiles.NotificationProfile
 
 class NotificationProfilesViewModel(private val repository: NotificationProfilesRepository) : ViewModel() {
 
-  fun getProfiles(): Observable<List<NotificationProfile>> {
+  fun getProfiles(): Flowable<List<NotificationProfile>> {
     return repository.getProfiles()
       .observeOn(AndroidSchedulers.mainThread())
   }

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationViewModel.java

@@ -52,7 +52,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
 
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
 import io.reactivex.rxjava3.core.BackpressureStrategy;
@@ -436,10 +435,10 @@ public class ConversationViewModel extends ViewModel {
   }
 
   @NonNull LiveData<Optional<NotificationProfile>> getActiveNotificationProfile() {
-    final Observable<Optional<NotificationProfile>> activeProfile = Observable.combineLatest(Observable.interval(0, 30, TimeUnit.SECONDS), notificationProfilesRepository.getProfiles(), (interval, profiles) -> profiles)
-                                                                              .map(profiles -> Optional.ofNullable(NotificationProfiles.getActiveProfile(profiles)));
+    Flowable<Optional<NotificationProfile>> activeProfile = notificationProfilesRepository.getProfiles()
+                                                                                          .map(profiles -> Optional.ofNullable(NotificationProfiles.getActiveProfile(profiles)));
 
-    return LiveDataReactiveStreams.fromPublisher(activeProfile.toFlowable(BackpressureStrategy.LATEST));
+    return LiveDataReactiveStreams.fromPublisher(activeProfile);
   }
 
   @NonNull

app/src/main/java/org/thoughtcrime/securesms/conversationlist/ConversationListFragment.java

@@ -79,6 +79,7 @@ import org.greenrobot.eventbus.Subscribe;
 import org.greenrobot.eventbus.ThreadMode;
 import org.signal.core.util.DimensionUnit;
 import org.signal.core.util.Stopwatch;
+import org.signal.core.util.concurrent.LifecycleDisposable;
 import org.signal.core.util.concurrent.SignalExecutors;
 import org.signal.core.util.concurrent.SimpleTask;
 import org.signal.core.util.logging.Log;
@@ -169,7 +170,6 @@ import org.thoughtcrime.securesms.util.BottomSheetUtil;
 import org.thoughtcrime.securesms.util.CachedInflater;
 import org.thoughtcrime.securesms.util.ConversationUtil;
 import org.thoughtcrime.securesms.util.FeatureFlags;
-import org.signal.core.util.concurrent.LifecycleDisposable;
 import org.thoughtcrime.securesms.util.PlayStoreUtil;
 import org.thoughtcrime.securesms.util.ServiceUtil;
 import org.thoughtcrime.securesms.util.SignalLocalMetrics;

app/src/main/java/org/thoughtcrime/securesms/conversationlist/ConversationListViewModel.kt

@@ -2,6 +2,7 @@ package org.thoughtcrime.securesms.conversationlist
 
 import androidx.lifecycle.ViewModel
 import androidx.lifecycle.ViewModelProvider
+import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.BackpressureStrategy
 import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Observable
@@ -185,13 +186,9 @@ class ConversationListViewModel(
     megaphoneRepository.markVisible(visible.event)
   }
 
-  fun getNotificationProfiles(): Observable<List<NotificationProfile>> {
-    return Observable
-      .combineLatest(
-        Observable.interval(0, 30, TimeUnit.SECONDS),
-        notificationProfilesRepository.getProfiles()
-      ) { _, profiles -> profiles }
-      .distinctUntilChanged()
+  fun getNotificationProfiles(): Flowable<List<NotificationProfile>> {
+    return notificationProfilesRepository.getProfiles()
+      .observeOn(AndroidSchedulers.mainThread())
   }
 
   private fun setSelection(newSelection: Collection<Conversation>) {

app/src/main/java/org/thoughtcrime/securesms/database/RxDatabaseObserver.kt

@@ -5,6 +5,7 @@ import io.reactivex.rxjava3.core.Emitter
 import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.schedulers.Schedulers
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import java.util.concurrent.TimeUnit
 
 /**
  * Provide a shared Rx interface to listen to database updates and ensure listeners
@@ -13,13 +14,28 @@ import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 object RxDatabaseObserver {
 
   val conversationList: Flowable<Unit> by lazy { conversationListFlowable() }
+  val notificationProfiles: Flowable<Unit> by lazy { notificationProfilesFlowable() }
 
   private fun conversationListFlowable(): Flowable<Unit> {
+    return databaseFlowable { listener ->
+      ApplicationDependencies.getDatabaseObserver().registerConversationListObserver(listener)
+    }
+  }
+
+  @Suppress("RedundantUnitExpression")
+  private fun notificationProfilesFlowable(): Flowable<Unit> {
+    return Flowable.combineLatest(
+      Flowable.interval(0, 30, TimeUnit.SECONDS),
+      databaseFlowable { ApplicationDependencies.getDatabaseObserver().registerNotificationProfileObserver(it) }
+    ) { _, _ -> Unit }
+  }
+
+  private fun databaseFlowable(registerObserver: (RxObserver) -> Unit): Flowable<Unit> {
     val flowable = Flowable.create(
       {
         val listener = RxObserver(it)
 
-        ApplicationDependencies.getDatabaseObserver().registerConversationListObserver(listener)
+        registerObserver(listener)
         it.setCancellable { ApplicationDependencies.getDatabaseObserver().unregisterObserver(listener) }
 
         listener.prime()