Fix possible RxStore memory leak.

app/src/main/java/org/thoughtcrime/securesms/badges/gifts/viewgift/received/ViewReceivedGiftViewModel.kt

@@ -70,6 +70,7 @@ class ViewReceivedGiftViewModel(
 
   override fun onCleared() {
     disposables.dispose()
+    store.dispose()
   }
 
   fun setChecked(isChecked: Boolean) {

app/src/main/java/org/thoughtcrime/securesms/badges/gifts/viewgift/sent/ViewSentGiftViewModel.kt

@@ -38,6 +38,7 @@ class ViewSentGiftViewModel(
 
   override fun onCleared() {
     disposables.dispose()
+    store.dispose()
   }
 
   class Factory(

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/internal/donor/DonorErrorConfigurationViewModel.kt

@@ -62,6 +62,7 @@ class DonorErrorConfigurationViewModel : ViewModel() {
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   fun setSelectedBadge(badgeIndex: Int) {

app/src/main/java/org/thoughtcrime/securesms/contacts/ContactChipViewModel.kt

@@ -35,6 +35,7 @@ class ContactChipViewModel : ViewModel() {
     disposables.clear()
     disposableMap.values.forEach { it.dispose() }
     disposableMap.clear()
+    store.dispose()
   }
 
   fun add(selectedContact: SelectedContact) {

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

@@ -59,6 +59,7 @@ import io.reactivex.rxjava3.core.BackpressureStrategy;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.processors.PublishProcessor;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import io.reactivex.rxjava3.subjects.BehaviorSubject;
@@ -135,10 +136,12 @@ public class ConversationViewModel extends ViewModel {
         .map(Recipient::resolved)
         .subscribe(recipientCache);
 
-    conversationStateStore.update(Observable.combineLatest(recipientId.distinctUntilChanged(), conversationStateTick, (id, tick) -> id)
+    Disposable disposable = conversationStateStore.update(Observable.combineLatest(recipientId.distinctUntilChanged(), conversationStateTick, (id, tick) -> id)
-                                            .switchMap(conversationRepository::getSecurityInfo)
+                                                                    .switchMap(conversationRepository::getSecurityInfo)
-                                            .toFlowable(BackpressureStrategy.LATEST),
+                                                                    .toFlowable(BackpressureStrategy.LATEST),
-                                  (securityInfo, state) -> state.withSecurityInfo(securityInfo));
+                                                          (securityInfo, state) -> state.withSecurityInfo(securityInfo));
+
+    disposables.add(disposable);
 
     BehaviorSubject<ConversationData> conversationMetadata = BehaviorSubject.create();
 
@@ -435,6 +438,7 @@ public class ConversationViewModel extends ViewModel {
     ApplicationDependencies.getDatabaseObserver().unregisterObserver(messageUpdateObserver);
     ApplicationDependencies.getDatabaseObserver().unregisterObserver(messageInsertObserver);
     disposables.clear();
+    conversationStateStore.dispose();
     EventBus.getDefault().unregister(this);
   }
 

app/src/main/java/org/thoughtcrime/securesms/conversation/drafts/DraftViewModel.kt

@@ -33,6 +33,10 @@ class DraftViewModel @JvmOverloads constructor(
   val voiceNoteDraft: Draft?
     get() = store.state.voiceNoteDraft
 
+  override fun onCleared() {
+    store.dispose()
+  }
+
   fun setThreadId(threadId: Long) {
     store.update { it.copy(threadId = threadId) }
   }

app/src/main/java/org/thoughtcrime/securesms/mediapreview/MediaPreviewV2ViewModel.kt

@@ -41,5 +41,6 @@ class MediaPreviewV2ViewModel : ViewModel() {
 
   override fun onCleared() {
     disposables.dispose()
+    store.dispose()
   }
 }

app/src/main/java/org/thoughtcrime/securesms/mediasend/v2/capture/MediaCaptureViewModel.kt

@@ -26,6 +26,10 @@ class MediaCaptureViewModel(private val repository: MediaCaptureRepository) : Vi
     }
   }
 
+  override fun onCleared() {
+    store.dispose()
+  }
+
   fun onImageCaptured(data: ByteArray, width: Int, height: Int) {
     repository.renderImageToMedia(data, width, height, this::onMediaRendered, this::onMediaRenderFailed)
   }

app/src/main/java/org/thoughtcrime/securesms/profiles/edit/pnp/WhoCanSeeMyPhoneNumberViewModel.kt

@@ -29,5 +29,6 @@ class WhoCanSeeMyPhoneNumberViewModel : ViewModel() {
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 }

app/src/main/java/org/thoughtcrime/securesms/profiles/manage/UsernameEditViewModel.java

@@ -65,6 +65,7 @@ class UsernameEditViewModel extends ViewModel {
   protected void onCleared() {
     super.onCleared();
     disposables.clear();
+    uiState.dispose();
   }
 
   void onNicknameUpdated(@NonNull String nickname) {

app/src/main/java/org/thoughtcrime/securesms/safety/SafetyNumberBottomSheetViewModel.kt

@@ -66,6 +66,8 @@ class SafetyNumberBottomSheetViewModel(
 
   override fun onCleared() {
     disposables.clear()
+    destinationStore.dispose()
+    store.dispose()
   }
 
   fun getIdentityRecord(recipientId: RecipientId): Maybe<IdentityRecord> {

app/src/main/java/org/thoughtcrime/securesms/sharing/v2/ShareViewModel.kt

@@ -72,6 +72,7 @@ class ShareViewModel(
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   private fun moveToFailedState(throwable: Throwable? = null) {

app/src/main/java/org/thoughtcrime/securesms/stories/settings/privacy/ChooseInitialMyStoryMembershipViewModel.kt

@@ -31,6 +31,7 @@ class ChooseInitialMyStoryMembershipViewModel @JvmOverloads constructor(
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   fun select(selection: DistributionListPrivacyMode): Single<DistributionListPrivacyMode> {

app/src/main/java/org/thoughtcrime/securesms/stories/settings/story/StoriesPrivacySettingsViewModel.kt

@@ -59,13 +59,14 @@ class StoriesPrivacySettingsViewModel : ViewModel() {
 
     pagingController.set(observablePagedData.controller)
 
-    store.update(observablePagedData.data.toFlowable(BackpressureStrategy.LATEST)) { data, state ->
+    disposables += store.update(observablePagedData.data.toFlowable(BackpressureStrategy.LATEST)) { data, state ->
       state.copy(storyContactItems = data)
     }
   }
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   fun setStoriesEnabled(isEnabled: Boolean) {

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/StoryViewerViewModel.kt

@@ -143,6 +143,7 @@ class StoryViewerViewModel(
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   fun setSelectedPage(page: Int) {

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/StoryVolumeViewModel.kt

@@ -10,6 +10,10 @@ class StoryVolumeViewModel : ViewModel() {
   val state: Flowable<StoryVolumeState> = store.stateFlowable
   val snapshot: StoryVolumeState get() = store.state
 
+  override fun onCleared() {
+    store.dispose()
+  }
+
   fun mute() {
     store.update { it.copy(isMuted = true) }
   }

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/info/StoryInfoViewModel.kt

@@ -75,6 +75,7 @@ class StoryInfoViewModel(storyId: Long, repository: StoryInfoRepository = StoryI
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   class Factory(private val storyId: Long) : ViewModelProvider.Factory {

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/page/StoryViewerPageViewModel.kt

@@ -99,6 +99,7 @@ class StoryViewerPageViewModel(
   override fun onCleared() {
     disposables.clear()
     storyCache.clear()
+    store.dispose()
   }
 
   fun hideStory(): Completable {

app/src/main/java/org/thoughtcrime/securesms/stories/viewer/reply/group/StoryGroupReplyViewModel.kt

@@ -51,6 +51,7 @@ class StoryGroupReplyViewModel(storyId: Long, repository: StoryGroupReplyReposit
 
   override fun onCleared() {
     disposables.clear()
+    store.dispose()
   }
 
   class Factory(private val storyId: Long, private val repository: StoryGroupReplyRepository) : ViewModelProvider.Factory {

app/src/main/java/org/thoughtcrime/securesms/util/rx/RxStore.kt

@@ -1,8 +1,10 @@
 package org.thoughtcrime.securesms.util.rx
 
+import androidx.annotation.CheckResult
 import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Scheduler
 import io.reactivex.rxjava3.disposables.Disposable
+import io.reactivex.rxjava3.kotlin.plusAssign
 import io.reactivex.rxjava3.processors.BehaviorProcessor
 import io.reactivex.rxjava3.schedulers.Schedulers
 import io.reactivex.rxjava3.subjects.PublishSubject
@@ -10,11 +12,14 @@ import io.reactivex.rxjava3.subjects.PublishSubject
 /**
  * Rx replacement for Store.
  * Actions are run on the computation thread by default.
+ *
+ * This class is disposable, and should be explicitly disposed of in a ViewModel's  onCleared method
+ * to prevent memory leaks. Disposing instances of this class is a terminal action.
  */
 class RxStore<T : Any>(
   defaultValue: T,
   scheduler: Scheduler = Schedulers.computation()
-) {
+) : Disposable {
 
   private val behaviorProcessor = BehaviorProcessor.createDefault(defaultValue)
   private val actionSubject = PublishSubject.create<(T) -> T>().toSerialized()
@@ -22,20 +27,30 @@ class RxStore<T : Any>(
   val state: T get() = behaviorProcessor.value!!
   val stateFlowable: Flowable<T> = behaviorProcessor.onBackpressureLatest()
 
-  init {
+  val actionDisposable: Disposable = actionSubject
-    actionSubject
+    .observeOn(scheduler)
-      .observeOn(scheduler)
+    .scan(defaultValue) { v, f -> f(v) }
-      .scan(defaultValue) { v, f -> f(v) }
+    .subscribe { behaviorProcessor.onNext(it) }
-      .subscribe { behaviorProcessor.onNext(it) }
-  }
 
   fun update(transformer: (T) -> T) {
     actionSubject.onNext(transformer)
   }
 
-  fun <U> update(flowable: Flowable<U>, transformer: (U, T) -> T): Disposable {
+  @CheckResult
+  fun <U : Any> update(flowable: Flowable<U>, transformer: (U, T) -> T): Disposable {
     return flowable.subscribe {
       actionSubject.onNext { t -> transformer(it, t) }
     }
   }
+
+  /**
+   * Dispose of the underlying scan chain. This is terminal.
+   */
+  override fun dispose() {
+    actionDisposable.dispose()
+  }
+
+  override fun isDisposed(): Boolean {
+    return actionDisposable.isDisposed
+  }
 }

app/src/test/java/org/thoughtcrime/securesms/util/rx/RxStoreTest.kt

@@ -33,6 +33,7 @@ class RxStoreTest {
     // THEN
     subscriber.assertValueAt(0, 1)
     subscriber.assertNotComplete()
+    testSubject.dispose()
   }
 
   @Test
@@ -50,6 +51,7 @@ class RxStoreTest {
     subscriber.assertValueAt(0, 1)
     subscriber.assertValueAt(1, 2)
     subscriber.assertNotComplete()
+    testSubject.dispose()
   }
 
   @Test
@@ -66,5 +68,6 @@ class RxStoreTest {
     // THEN
     subscriber.assertValueAt(0, 2)
     subscriber.assertNotComplete()
+    testSubject.dispose()
   }
 }