Properly support group calls in CFv2.

app/src/main/java/org/thoughtcrime/securesms/conversation/v2/ConversationFragment.kt

@@ -197,6 +197,7 @@ import org.thoughtcrime.securesms.database.model.StickerRecord
 import org.thoughtcrime.securesms.database.model.databaseprotos.BodyRangeList
 import org.thoughtcrime.securesms.databinding.V2ConversationFragmentBinding
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.events.GroupCallPeekEvent
 import org.thoughtcrime.securesms.giph.mp4.GiphyMp4ItemDecoration
 import org.thoughtcrime.securesms.giph.mp4.GiphyMp4PlaybackController
 import org.thoughtcrime.securesms.giph.mp4.GiphyMp4PlaybackPolicy
@@ -388,11 +389,9 @@ class ConversationFragment :
     )
   }
 
-  private val groupCallViewModel: ConversationGroupCallViewModel by viewModels(
-    factoryProducer = {
-      ConversationGroupCallViewModel.Factory(args.threadId, conversationRecipientRepository)
-    }
-  )
+  private val groupCallViewModel: ConversationGroupCallViewModel by viewModel {
+    ConversationGroupCallViewModel(conversationRecipientRepository)
+  }
 
   private val conversationGroupViewModel: ConversationGroupViewModel by viewModels(
     factoryProducer = {
@@ -801,7 +800,6 @@ class ConversationFragment :
 
     attachmentManager = AttachmentManager(requireContext(), requireView(), AttachmentManagerListener())
 
-    EventBus.getDefault().registerForLifecycle(groupCallViewModel, viewLifecycleOwner)
     viewLifecycleOwner.lifecycle.addObserver(LastScrolledPositionUpdater(adapter, layoutManager, viewModel))
 
     disposables += viewModel.recipient
@@ -1278,16 +1276,14 @@ class ConversationFragment :
       handleVideoCall()
     }
 
-    disposables += groupCallViewModel.hasActiveGroupCall.subscribeBy(onNext = {
-      invalidateOptionsMenu()
-      binding.conversationGroupCallJoin.visible = it
-    })
-
-    disposables += groupCallViewModel.hasCapacity.subscribeBy(onNext = {
-      binding.conversationGroupCallJoin.setText(
-        if (it) R.string.ConversationActivity_join else R.string.ConversationActivity_full
-      )
-    })
+    disposables += groupCallViewModel
+      .state
+      .distinctUntilChanged()
+      .subscribeBy {
+        binding.conversationGroupCallJoin.visible = it.ongoingCall
+        binding.conversationGroupCallJoin.setText(if (it.hasCapacity) R.string.ConversationActivity_join else R.string.ConversationActivity_full)
+        invalidateOptionsMenu()
+      }
   }
 
   private fun handleVideoCall() {
@@ -1297,7 +1293,7 @@ class ConversationFragment :
       return
     }
 
-    val hasActiveGroupCall: Single<Boolean> = groupCallViewModel.hasActiveGroupCall.firstOrError()
+    val hasActiveGroupCall: Single<Boolean> = groupCallViewModel.state.map { it.ongoingCall }.firstOrError()
     val isNonAdminInAnnouncementGroup: Boolean = conversationGroupViewModel.isNonAdminInAnnouncementGroup()
     val cannotCreateGroupCall = hasActiveGroupCall.map { active ->
       recipient to (recipient.isPushV2Group && !active && isNonAdminInAnnouncementGroup)
@@ -2865,7 +2861,7 @@ class ConversationFragment :
         isActiveGroup = recipient?.isActiveGroup == true,
         isActiveV2Group = recipient?.let { it.isActiveGroup && it.isPushV2Group } == true,
         isInActiveGroup = recipient?.isActiveGroup == false,
-        hasActiveGroupCall = groupCallViewModel.hasActiveGroupCallSnapshot,
+        hasActiveGroupCall = groupCallViewModel.hasOngoingGroupCallSnapshot,
         distributionType = args.distributionType,
         threadId = args.threadId,
         isInMessageRequest = viewModel.hasMessageRequestState,
@@ -3871,6 +3867,11 @@ class ConversationFragment :
     }
   }
 
+  @Subscribe(threadMode = ThreadMode.MAIN, sticky = true)
+  fun onGroupCallPeekEvent(groupCallPeekEvent: GroupCallPeekEvent) {
+    groupCallViewModel.onGroupCallPeekEvent(groupCallPeekEvent)
+  }
+
   //endregion
 
   private inner class SearchEventListener : ConversationSearchBottomBar.EventListener {

app/src/main/java/org/thoughtcrime/securesms/conversation/v2/groups/ConversationGroupCallState.kt

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.conversation.v2.groups
+
+import org.thoughtcrime.securesms.recipients.RecipientId
+
+/** State of a group call used solely within rendering UX/UI in the conversation */
+data class ConversationGroupCallState(
+  val recipientId: RecipientId? = null,
+  val activeV2Group: Boolean = false,
+  val ongoingCall: Boolean = false,
+  val hasCapacity: Boolean = false
+)

app/src/main/java/org/thoughtcrime/securesms/conversation/v2/groups/ConversationGroupCallViewModel.kt

@@ -1,31 +1,23 @@
 package org.thoughtcrime.securesms.conversation.v2.groups
 
 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.Single
+import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.disposables.CompositeDisposable
-import io.reactivex.rxjava3.kotlin.plusAssign
+import io.reactivex.rxjava3.kotlin.addTo
 import io.reactivex.rxjava3.kotlin.subscribeBy
 import io.reactivex.rxjava3.processors.PublishProcessor
 import io.reactivex.rxjava3.schedulers.Schedulers
-import io.reactivex.rxjava3.subjects.BehaviorSubject
-import io.reactivex.rxjava3.subjects.Subject
-import org.greenrobot.eventbus.Subscribe
-import org.greenrobot.eventbus.ThreadMode
 import org.signal.core.util.logging.Log
 import org.thoughtcrime.securesms.conversation.v2.ConversationRecipientRepository
-import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.events.GroupCallPeekEvent
-import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.util.rx.RxStore
 
 /**
  * ViewModel which manages state associated with group calls.
  */
 class ConversationGroupCallViewModel(
-  threadId: Long,
   recipientRepository: ConversationRecipientRepository
 ) : ViewModel() {
 
@@ -33,85 +25,62 @@ class ConversationGroupCallViewModel(
     private val TAG = Log.tag(ConversationGroupCallViewModel::class.java)
   }
 
-  private val _isGroupActive: Subject<Boolean> = BehaviorSubject.createDefault(false)
-  private val _hasOngoingGroupCall: Subject<Boolean> = BehaviorSubject.createDefault(false)
-  private val _hasCapacity: Subject<Boolean> = BehaviorSubject.createDefault(false)
-  private val _hasActiveGroupCall: BehaviorSubject<Boolean> = BehaviorSubject.create()
-  private val _recipient: BehaviorSubject<Recipient> = BehaviorSubject.create()
-  private val _groupCallPeekEventProcessor: PublishProcessor<GroupCallPeekEvent> = PublishProcessor.create()
-  private val _peekRequestProcessor: PublishProcessor<Unit> = PublishProcessor.create()
   private val disposables = CompositeDisposable()
+  private val store = RxStore(ConversationGroupCallState()).addTo(disposables)
+  private val forcePeek = PublishProcessor.create<Unit>()
 
-  val hasActiveGroupCall: Observable<Boolean> = _hasActiveGroupCall.observeOn(AndroidSchedulers.mainThread())
-  val hasCapacity: Observable<Boolean> = _hasCapacity.observeOn(AndroidSchedulers.mainThread())
+  val state: Flowable<ConversationGroupCallState> = store.stateFlowable.onBackpressureLatest().observeOn(AndroidSchedulers.mainThread())
 
-  val hasActiveGroupCallSnapshot: Boolean
-    get() = _hasActiveGroupCall.value == true
+  val hasOngoingGroupCallSnapshot: Boolean
+    get() = store.state.ongoingCall
 
   init {
-    disposables += Observable
-      .combineLatest(_isGroupActive, _hasActiveGroupCall) { a, b -> a && b }
-      .subscribeBy(onNext = _hasActiveGroupCall::onNext)
+    recipientRepository
+      .conversationRecipient
+      .subscribeBy { recipient ->
+        store.update { s: ConversationGroupCallState ->
+          val activeV2Group = recipient.isPushV2Group && recipient.isActiveGroup
+          s.copy(
+            recipientId = recipient.id,
+            activeV2Group = activeV2Group,
+            ongoingCall = if (activeV2Group && s.recipientId == recipient.id) s.ongoingCall else false,
+            hasCapacity = if (activeV2Group && s.recipientId == recipient.id) s.hasCapacity else false
+          )
+        }
+      }
+      .addTo(disposables)
 
-    disposables += Single
-      .fromCallable { SignalDatabase.threads.getRecipientForThreadId(threadId)!! }
+    val filteredState = store.stateFlowable
+      .filter { it.recipientId != null }
+      .distinctUntilChanged { s -> s.activeV2Group }
+
+    Flowable.combineLatest(forcePeek, filteredState) { _, s -> s }
       .subscribeOn(Schedulers.io())
-      .filter { it.isPushV2Group }
-      .flatMapObservable { Recipient.live(it.id).observable() }
-      .subscribeBy(onNext = _recipient::onNext)
-
-    disposables += _recipient
-      .map { it.isActiveGroup }
-      .distinctUntilChanged()
-      .subscribeBy(onNext = _isGroupActive::onNext)
-
-    disposables += _recipient
-      .firstOrError()
-      .subscribeBy(onSuccess = {
-        peekGroupCall()
-      })
-
-    disposables += _groupCallPeekEventProcessor
       .onBackpressureLatest()
-      .switchMap { event ->
-        _recipient.firstElement().map { it.id }.filter { it == event.groupRecipientId }.map { event }.toFlowable()
+      .subscribeBy { s: ConversationGroupCallState ->
+        if (s.recipientId != null && s.activeV2Group) {
+          Log.i(TAG, "Peek call for ${s.recipientId}")
+          ApplicationDependencies.getSignalCallManager().peekGroupCall(s.recipientId)
+        }
       }
-      .subscribeBy(onNext = {
-        Log.i(TAG, "update UI with call event: ongoing call: " + it.isOngoing + " hasCapacity: " + it.callHasCapacity())
-        _hasOngoingGroupCall.onNext(it.isOngoing)
-        _hasCapacity.onNext(it.callHasCapacity())
-      })
-
-    disposables += _peekRequestProcessor
-      .onBackpressureLatest()
-      .switchMap {
-        _recipient.firstOrError().map { it.id }.toFlowable()
-      }
-      .subscribeBy(onNext = { recipientId ->
-        Log.i(TAG, "peek call for $recipientId")
-        ApplicationDependencies.getSignalCallManager().peekGroupCall(recipientId)
-      })
+      .addTo(disposables)
   }
 
   override fun onCleared() {
     disposables.clear()
   }
 
-  @Subscribe(threadMode = ThreadMode.MAIN, sticky = true)
-  fun onGroupCallPeekEvent(groupCallPeekEvent: GroupCallPeekEvent) {
-    _groupCallPeekEventProcessor.onNext(groupCallPeekEvent)
+  fun onGroupCallPeekEvent(event: GroupCallPeekEvent) {
+    store.update { s: ConversationGroupCallState ->
+      if (s.recipientId != null && event.groupRecipientId == s.recipientId) {
+        s.copy(ongoingCall = event.isOngoing, hasCapacity = event.callHasCapacity())
+      } else {
+        s
+      }
+    }
   }
 
   fun peekGroupCall() {
-    _peekRequestProcessor.onNext(Unit)
-  }
-
-  class Factory(
-    private val threadId: Long,
-    private val recipientRepository: ConversationRecipientRepository
-  ) : ViewModelProvider.Factory {
-    override fun <T : ViewModel> create(modelClass: Class<T>): T {
-      return modelClass.cast(ConversationGroupCallViewModel(threadId, recipientRepository)) as T
-    }
+    forcePeek.onNext(Unit)
   }
 }