Reimplement MessageRequestViewModel for CFV2.

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

@@ -137,23 +137,31 @@ class ConversationFragment : LoggingFragment(R.layout.v2_conversation_fragment)
     ConversationIntents.Args.from(requireArguments())
   }
 
+  private val conversationRecipientRepository: ConversationRecipientRepository by lazy {
+    ConversationRecipientRepository(args.threadId)
+  }
+
   private val disposables = LifecycleDisposable()
   private val binding by ViewBinderDelegate(V2ConversationFragmentBinding::bind)
   private val viewModel: ConversationViewModel by viewModels(
     factoryProducer = {
-      ConversationViewModel.Factory(args, ConversationRepository(requireContext()))
+      ConversationViewModel.Factory(
+        args,
+        ConversationRepository(requireContext()),
+        conversationRecipientRepository
+      )
     }
   )
 
   private val groupCallViewModel: ConversationGroupCallViewModel by viewModels(
     factoryProducer = {
-      ConversationGroupCallViewModel.Factory(args.threadId)
+      ConversationGroupCallViewModel.Factory(args.threadId, conversationRecipientRepository)
     }
   )
 
   private val conversationGroupViewModel: ConversationGroupViewModel by viewModels(
     factoryProducer = {
-      ConversationGroupViewModel.Factory(args.threadId)
+      ConversationGroupViewModel.Factory(args.threadId, conversationRecipientRepository)
     }
   )
 

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

@@ -0,0 +1,24 @@
+package org.thoughtcrime.securesms.conversation.v2
+
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.schedulers.Schedulers
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.recipients.Recipient
+
+class ConversationRecipientRepository(threadId: Long) {
+
+  val conversationRecipient: Observable<Recipient> by lazy {
+    val threadRecipientId = Single.fromCallable {
+      SignalDatabase.threads.getRecipientIdForThreadId(threadId)!!
+    }
+
+    threadRecipientId
+      .flatMapObservable { Recipient.observable(it) }
+      .subscribeOn(Schedulers.io())
+      .observeOn(Schedulers.io())
+      .replay(1)
+      .refCount()
+      .observeOn(Schedulers.io())
+  }
+}

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

@@ -4,7 +4,6 @@ import android.content.Context
 import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Observable
 import io.reactivex.rxjava3.core.Single
-import io.reactivex.rxjava3.kotlin.subscribeBy
 import io.reactivex.rxjava3.schedulers.Schedulers
 import org.signal.core.util.concurrent.SignalExecutors
 import org.signal.paging.PagedData
@@ -25,28 +24,6 @@ class ConversationRepository(context: Context) {
   private val applicationContext = context.applicationContext
   private val oldConversationRepository = org.thoughtcrime.securesms.conversation.ConversationRepository()
 
-  /**
-   * Observes the recipient tied to the given thread id, returning an error if
-   * the thread id does not exist or somehow does not have a recipient attached to it.
-   */
-  fun observeRecipientForThread(threadId: Long): Observable<Recipient> {
-    return Observable.create { emitter ->
-      val recipientId = SignalDatabase.threads.getRecipientIdForThreadId(threadId)
-
-      if (recipientId != null) {
-        val disposable = Recipient.live(recipientId).observable()
-          .subscribeOn(Schedulers.io())
-          .subscribeBy(onNext = emitter::onNext)
-
-        emitter.setCancellable {
-          disposable.dispose()
-        }
-      } else {
-        emitter.onError(Exception("Thread $threadId does not exist."))
-      }
-    }.subscribeOn(Schedulers.io())
-  }
-
   /**
    * Loads the details necessary to display the conversation thread.
    */

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

@@ -34,7 +34,8 @@ import org.thoughtcrime.securesms.wallpaper.ChatWallpaper
 class ConversationViewModel(
   private val threadId: Long,
   requestedStartingPosition: Int,
-  private val repository: ConversationRepository
+  private val repository: ConversationRepository,
+  recipientRepository: ConversationRecipientRepository
 ) : ViewModel() {
 
   private val disposables = CompositeDisposable()
@@ -67,7 +68,8 @@ class ConversationViewModel(
     get() = _recipient.value?.wallpaper
 
   init {
-    disposables += repository.observeRecipientForThread(threadId)
+    disposables += recipientRepository
+      .conversationRecipient
       .subscribeBy(onNext = {
         _recipient.onNext(it)
       })
@@ -149,10 +151,18 @@ class ConversationViewModel(
 
   class Factory(
     private val args: Args,
-    private val repository: ConversationRepository
+    private val repository: ConversationRepository,
+    private val recipientRepository: ConversationRecipientRepository
   ) : ViewModelProvider.Factory {
     override fun <T : ViewModel> create(modelClass: Class<T>): T {
-      return modelClass.cast(ConversationViewModel(args.threadId, args.startingPosition, repository)) as T
+      return modelClass.cast(
+        ConversationViewModel(
+          args.threadId,
+          args.startingPosition,
+          repository,
+          recipientRepository
+        )
+      ) as T
     }
   }
 }

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

@@ -0,0 +1,180 @@
+package org.thoughtcrime.securesms.conversation.v2
+
+import androidx.lifecycle.ViewModel
+import androidx.lifecycle.ViewModelProvider
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.disposables.CompositeDisposable
+import io.reactivex.rxjava3.kotlin.plusAssign
+import io.reactivex.rxjava3.kotlin.subscribeBy
+import io.reactivex.rxjava3.subjects.BehaviorSubject
+import io.reactivex.rxjava3.subjects.PublishSubject
+import org.signal.core.util.concurrent.subscribeWithSubject
+import org.thoughtcrime.securesms.groups.ui.GroupChangeFailureReason
+import org.thoughtcrime.securesms.messagerequests.GroupInfo
+import org.thoughtcrime.securesms.messagerequests.MessageRequestRepository
+import org.thoughtcrime.securesms.messagerequests.MessageRequestState
+import org.thoughtcrime.securesms.messagerequests.MessageRequestViewModel.MessageData
+import org.thoughtcrime.securesms.messagerequests.MessageRequestViewModel.RecipientInfo
+import org.thoughtcrime.securesms.messagerequests.MessageRequestViewModel.RequestReviewDisplayState
+import org.thoughtcrime.securesms.messagerequests.MessageRequestViewModel.Status
+import org.thoughtcrime.securesms.profiles.spoofing.ReviewUtil
+
+/**
+ * MessageRequestViewModel for ConversationFragment V2
+ */
+class MessageRequestViewModel(
+  private val threadId: Long,
+  private val recipientRepository: ConversationRecipientRepository,
+  private val messageRequestRepository: MessageRequestRepository
+) : ViewModel() {
+
+  private val disposables = CompositeDisposable()
+
+  private val statusSubject = PublishSubject.create<Status>()
+  val status: Observable<Status> = statusSubject
+
+  private val failureSubject = PublishSubject.create<GroupChangeFailureReason>()
+  val failure: Observable<GroupChangeFailureReason> = failureSubject
+
+  private val groupInfo: Observable<GroupInfo> = recipientRepository
+    .conversationRecipient
+    .flatMap { recipient ->
+      Single.create { emitter ->
+        messageRequestRepository.getGroupInfo(recipient.id, emitter::onSuccess)
+      }.toObservable()
+    }
+
+  private val groups: Observable<List<String>> = recipientRepository
+    .conversationRecipient
+    .flatMap { recipient ->
+      Single.create<List<String>> { emitter ->
+        messageRequestRepository.getGroups(recipient.id, emitter::onSuccess)
+      }.toObservable()
+    }
+
+  private val messageDataSubject: BehaviorSubject<MessageData> = recipientRepository.conversationRecipient.map {
+    val state = messageRequestRepository.getMessageRequestState(it, threadId)
+    MessageData(it, state)
+  }.subscribeWithSubject(BehaviorSubject.create(), disposables)
+
+  private val requestReviewDisplayStateSubject: BehaviorSubject<RequestReviewDisplayState> = messageDataSubject.map { holder ->
+    if (holder.messageState == MessageRequestState.INDIVIDUAL) {
+      if (ReviewUtil.isRecipientReviewSuggested(holder.recipient.id)) {
+        RequestReviewDisplayState.SHOWN
+      } else {
+        RequestReviewDisplayState.HIDDEN
+      }
+    } else {
+      RequestReviewDisplayState.NONE
+    }
+  }.subscribeWithSubject(BehaviorSubject.create(), disposables)
+
+  val recipientInfo: Observable<RecipientInfo> = Observable.combineLatest(
+    recipientRepository.conversationRecipient,
+    groupInfo,
+    groups,
+    messageDataSubject.map { it.messageState },
+    ::RecipientInfo
+  )
+
+  override fun onCleared() {
+    disposables.clear()
+  }
+
+  fun shouldShowMessageRequest(): Boolean {
+    val messageData = messageDataSubject.value
+    return messageData != null && messageData.messageState != MessageRequestState.NONE
+  }
+
+  fun onAccept() {
+    statusSubject.onNext(Status.ACCEPTING)
+    disposables += recipientRepository
+      .conversationRecipient
+      .firstOrError()
+      .map { it.id }
+      .subscribeBy { recipientId ->
+        messageRequestRepository.acceptMessageRequest(
+          recipientId,
+          threadId,
+          { statusSubject.onNext(Status.ACCEPTED) },
+          this::onGroupChangeError
+        )
+      }
+  }
+  fun onDelete() {
+    statusSubject.onNext(Status.DELETING)
+    disposables += recipientRepository
+      .conversationRecipient
+      .firstOrError()
+      .map { it.id }
+      .subscribeBy { recipientId ->
+        messageRequestRepository.deleteMessageRequest(
+          recipientId,
+          threadId,
+          { statusSubject.onNext(Status.DELETED) },
+          this::onGroupChangeError
+        )
+      }
+  }
+  fun onBlock() {
+    statusSubject.onNext(Status.BLOCKING)
+    disposables += recipientRepository
+      .conversationRecipient
+      .firstOrError()
+      .map { it.id }
+      .subscribeBy { recipientId ->
+        messageRequestRepository.blockMessageRequest(
+          recipientId,
+          { statusSubject.onNext(Status.BLOCKED) },
+          this::onGroupChangeError
+        )
+      }
+  }
+  fun onUnblock() {
+    disposables += recipientRepository
+      .conversationRecipient
+      .firstOrError()
+      .map { it.id }
+      .subscribeBy { recipientId ->
+        messageRequestRepository.unblockAndAccept(
+          recipientId
+        ) { statusSubject.onNext(Status.ACCEPTED) }
+      }
+  }
+  fun onBlockAndReportSpam() {
+    disposables += recipientRepository
+      .conversationRecipient
+      .firstOrError()
+      .map { it.id }
+      .subscribeBy { recipientId ->
+        messageRequestRepository.blockAndReportSpamMessageRequest(
+          recipientId,
+          threadId,
+          { statusSubject.onNext(Status.BLOCKED_AND_REPORTED) },
+          this::onGroupChangeError
+        )
+      }
+  }
+
+  private fun onGroupChangeError(error: GroupChangeFailureReason) {
+    statusSubject.onNext(Status.IDLE)
+    failureSubject.onNext(error)
+  }
+
+  class Factory(
+    private val threadId: Long,
+    private val recipientRepository: ConversationRecipientRepository,
+    private val messageRequestRepository: MessageRequestRepository
+  ) : ViewModelProvider.Factory {
+    override fun <T : ViewModel> create(modelClass: Class<T>): T {
+      return modelClass.cast(
+        MessageRequestViewModel(
+          threadId,
+          recipientRepository,
+          messageRequestRepository
+        )
+      ) as T
+    }
+  }
+}

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

@@ -15,6 +15,7 @@ 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
@@ -23,7 +24,10 @@ import org.thoughtcrime.securesms.recipients.Recipient
 /**
  * ViewModel which manages state associated with group calls.
  */
-class ConversationGroupCallViewModel(threadId: Long) : ViewModel() {
+class ConversationGroupCallViewModel(
+  threadId: Long,
+  recipientRepository: ConversationRecipientRepository
+) : ViewModel() {
 
   companion object {
     private val TAG = Log.tag(ConversationGroupCallViewModel::class.java)
@@ -102,9 +106,12 @@ class ConversationGroupCallViewModel(threadId: Long) : ViewModel() {
     _peekRequestProcessor.onNext(Unit)
   }
 
-  class Factory(private val threadId: Long) : ViewModelProvider.Factory {
+  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)) as T
+      return modelClass.cast(ConversationGroupCallViewModel(threadId, recipientRepository)) as T
     }
   }
 }

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

@@ -8,9 +8,9 @@ import io.reactivex.rxjava3.core.Single
 import io.reactivex.rxjava3.disposables.CompositeDisposable
 import io.reactivex.rxjava3.kotlin.plusAssign
 import io.reactivex.rxjava3.kotlin.subscribeBy
-import io.reactivex.rxjava3.schedulers.Schedulers
 import io.reactivex.rxjava3.subjects.BehaviorSubject
 import io.reactivex.rxjava3.subjects.Subject
+import org.thoughtcrime.securesms.conversation.v2.ConversationRecipientRepository
 import org.thoughtcrime.securesms.database.GroupTable
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.database.model.GroupRecord
@@ -26,7 +26,8 @@ import org.thoughtcrime.securesms.recipients.RecipientId
  */
 class ConversationGroupViewModel(
   private val threadId: Long,
-  private val groupManagementRepository: GroupManagementRepository = GroupManagementRepository()
+  private val groupManagementRepository: GroupManagementRepository = GroupManagementRepository(),
+  private val recipientRepository: ConversationRecipientRepository
 ) : ViewModel() {
 
   private val disposables = CompositeDisposable()
@@ -39,11 +40,9 @@ class ConversationGroupViewModel(
   private val _reviewState: Subject<ConversationGroupReviewState> = BehaviorSubject.create()
 
   init {
-    disposables += Single
-      .fromCallable { SignalDatabase.threads.getRecipientForThreadId(threadId)!! }
-      .subscribeOn(Schedulers.io())
+    disposables += recipientRepository
+      .conversationRecipient
       .filter { it.isGroup }
-      .flatMapObservable { Recipient.observable(it.id) }
       .subscribeBy(onNext = _recipient::onNext)
 
     disposables += _recipient
@@ -115,9 +114,9 @@ class ConversationGroupViewModel(
     }
   }
 
-  class Factory(private val threadId: Long) : ViewModelProvider.Factory {
+  class Factory(private val threadId: Long, private val recipientRepository: ConversationRecipientRepository) : ViewModelProvider.Factory {
     override fun <T : ViewModel> create(modelClass: Class<T>): T {
-      return modelClass.cast(ConversationGroupViewModel(threadId)) as T
+      return modelClass.cast(ConversationGroupViewModel(threadId, recipientRepository = recipientRepository)) as T
     }
   }
 }