Mavericks 2: start replacing rx by flow

vector/src/main/java/im/vector/app/features/crypto/quads/SharedSecureStorageViewModel.kt

@@ -35,6 +35,7 @@ import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.platform.WaitingViewData
 import im.vector.app.core.resources.StringProvider
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import org.matrix.android.sdk.api.listeners.ProgressListener
@@ -42,6 +43,7 @@ import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.securestorage.IntegrityResult
 import org.matrix.android.sdk.api.session.securestorage.KeyInfoResult
 import org.matrix.android.sdk.api.session.securestorage.RawBytesKeySpec
+import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.crosssigning.toBase64NoPadding
 import org.matrix.android.sdk.rx.rx
 import timber.log.Timber
@@ -114,7 +116,7 @@ class SharedSecureStorageViewModel @AssistedInject constructor(
             }
         }
 
-        session.rx()
+        session.flow()
                 .liveUserCryptoDevices(session.myUserId)
                 .distinctUntilChanged()
                 .execute {

vector/src/main/java/im/vector/app/features/devtools/RoomDevToolViewModel.kt

@@ -40,6 +40,7 @@ import org.matrix.android.sdk.api.session.events.model.EventType
 import org.matrix.android.sdk.api.session.events.model.toModel
 import org.matrix.android.sdk.api.session.room.model.message.MessageContent
 import org.matrix.android.sdk.api.util.JsonDict
+import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.di.MoshiProvider
 import org.matrix.android.sdk.rx.rx
 
@@ -69,7 +70,7 @@ class RoomDevToolViewModel @AssistedInject constructor(
 
     init {
         session.getRoom(initialState.roomId)
-                ?.rx()
+                ?.flow()
                 ?.liveStateEvents(emptySet())
                 ?.execute { async ->
                     copy(stateEvents = async)

vector/src/main/java/im/vector/app/features/home/room/detail/RoomDetailViewModel.kt

@@ -18,7 +18,7 @@ package im.vector.app.features.home.room.detail
 
 import android.net.Uri
 import androidx.annotation.IdRes
-import androidx.lifecycle.viewModelScope
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -65,6 +65,10 @@ import io.reactivex.rxkotlin.subscribeBy
 import io.reactivex.schedulers.Schedulers
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import org.commonmark.parser.Parser
@@ -104,8 +108,9 @@ import org.matrix.android.sdk.api.session.room.timeline.getTextEditableContent
 import org.matrix.android.sdk.api.session.space.CreateSpaceParams
 import org.matrix.android.sdk.api.session.widgets.model.WidgetType
 import org.matrix.android.sdk.api.util.toOptional
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 import org.matrix.android.sdk.internal.crypto.model.event.WithHeldCode
-import org.matrix.android.sdk.rx.asObservable
 import org.matrix.android.sdk.rx.rx
 import org.matrix.android.sdk.rx.unwrap
 import timber.log.Timber
@@ -252,7 +257,7 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeActiveRoomWidgets() {
-        session.rx()
+        session.flow()
                 .liveRoomWidgets(
                         roomId = initialState.roomId,
                         widgetId = QueryStringValue.NoCondition
@@ -285,7 +290,7 @@ class RoomDetailViewModel @AssistedInject constructor(
         val queryParams = roomMemberQueryParams {
             this.userId = QueryStringValue.Equals(session.myUserId, QueryStringValue.Case.SENSITIVE)
         }
-        room.rx()
+        room.flow()
                 .liveRoomMembers(queryParams)
                 .map {
                     it.firstOrNull().toOptional()
@@ -1503,29 +1508,22 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeSyncState() {
-        session.rx()
+        session.flow()
                 .liveSyncState()
-                .subscribe { syncState ->
-                    setState {
-                        copy(syncState = syncState)
-                    }
+                .setOnEach { syncState ->
+                    copy(syncState = syncState)
                 }
-                .disposeOnClear()
 
         session.getSyncStatusLive()
-                .asObservable()
-                .subscribe { it ->
-                    if (it is SyncStatusService.Status.IncrementalSyncStatus) {
-                        setState {
-                            copy(incrementalSyncStatus = it)
-                        }
-                    }
+                .asFlow()
+                .filterIsInstance<SyncStatusService.Status.IncrementalSyncStatus>()
+                .setOnEach {
+                    copy(incrementalSyncStatus = it)
                 }
-                .disposeOnClear()
     }
 
     private fun observeRoomSummary() {
-        room.rx().liveRoomSummary()
+        room.flow().liveRoomSummary()
                 .unwrap()
                 .execute { async ->
                     copy(
@@ -1587,16 +1585,15 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.rx()
+        session.flow()
                 .liveRoomChangeMembershipState()
                 .map {
                     it[initialState.roomId] ?: ChangeMembershipState.Unknown
                 }
                 .distinctUntilChanged()
-                .subscribe {
-                    setState { copy(changeMembershipState = it) }
+                .setOnEach {
+                    copy(changeMembershipState = it)
                 }
-                .disposeOnClear()
     }
 
     private fun observeSummaryState() {

vector/src/main/java/im/vector/app/features/home/room/detail/timeline/action/MessageActionsViewModel.kt

@@ -36,6 +36,11 @@ import im.vector.app.features.html.VectorHtmlCompressor
 import im.vector.app.features.powerlevel.PowerLevelsObservableFactory
 import im.vector.app.features.reactions.data.EmojiDataSource
 import im.vector.app.features.settings.VectorPreferences
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.switchMap
 import org.matrix.android.sdk.api.extensions.orFalse
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.crypto.keysbackup.KeysBackupState
@@ -54,8 +59,9 @@ import org.matrix.android.sdk.api.session.room.send.SendState
 import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
 import org.matrix.android.sdk.api.session.room.timeline.getLastMessageContent
 import org.matrix.android.sdk.api.session.room.timeline.hasBeenEdited
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 import org.matrix.android.sdk.rx.rx
-import org.matrix.android.sdk.rx.unwrap
 import java.util.ArrayList
 
 /**
@@ -79,7 +85,7 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
         pillsPostProcessorFactory.create(initialState.roomId)
     }
 
-    private val eventIdObservable = BehaviorRelay.createDefault(initialState.eventId)
+    private val eventIdFlow = MutableStateFlow(initialState.eventId)
 
     @AssistedFactory
     interface Factory {
@@ -135,7 +141,7 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
 
     private fun observeEvent() {
         if (room == null) return
-        room.rx()
+        room.flow()
                 .liveTimelineEvent(initialState.eventId)
                 .unwrap()
                 .execute {
@@ -145,9 +151,9 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
 
     private fun observeReactions() {
         if (room == null) return
-        eventIdObservable
-                .switchMap { eventId ->
-                    room.rx()
+        eventIdFlow
+                .flatMapLatest { eventId ->
+                    room.flow()
                             .liveAnnotationSummary(eventId)
                             .map { annotations ->
                                 EmojiDataSource.quickEmojis.map { emoji ->
@@ -163,7 +169,7 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
     private fun observeTimelineEventState() {
         selectSubscribe(MessageActionState::timelineEvent, MessageActionState::actionPermissions) { timelineEvent, permissions ->
             val nonNullTimelineEvent = timelineEvent() ?: return@selectSubscribe
-            eventIdObservable.accept(nonNullTimelineEvent.eventId)
+            eventIdFlow.tryEmit(nonNullTimelineEvent.eventId)
             setState {
                 copy(
                         eventId = nonNullTimelineEvent.eventId,

vector/src/main/java/im/vector/app/features/home/room/list/RoomListViewModel.kt

@@ -17,12 +17,11 @@
 package im.vector.app.features.home.room.list
 
 import androidx.lifecycle.MutableLiveData
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
-import com.airbnb.mvrx.MvRxViewModelFactory
+import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.Success
 import com.airbnb.mvrx.ViewModelContext
 import im.vector.app.AppStateHandler
@@ -33,6 +32,8 @@ import im.vector.app.core.resources.StringProvider
 import im.vector.app.features.invite.AutoAcceptInvites
 import im.vector.app.features.settings.VectorPreferences
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.orFalse
 import org.matrix.android.sdk.api.query.QueryStringValue
@@ -41,7 +42,7 @@ import org.matrix.android.sdk.api.session.room.UpdatableLivePageResult
 import org.matrix.android.sdk.api.session.room.members.ChangeMembershipState
 import org.matrix.android.sdk.api.session.room.model.tag.RoomTag
 import org.matrix.android.sdk.api.session.room.state.isPublic
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 import javax.inject.Inject
 
@@ -72,11 +73,13 @@ class RoomListViewModel @Inject constructor(
          * If current space is null, will return orphan rooms only
          */
         ORPHANS_IF_SPACE_NULL,
+
         /**
          * Special case when we don't want to discriminate rooms when current space is null.
          * In this case return all.
          */
         ALL_IF_SPACE_NULL,
+
         /** Do not filter based on space*/
         NONE
     }
@@ -92,7 +95,7 @@ class RoomListViewModel @Inject constructor(
                     )
                 }
 
-        session.rx().liveUser(session.myUserId)
+        session.flow().liveUser(session.myUserId)
                 .map { it.getOrNull()?.getBestName() }
                 .distinctUntilChanged()
                 .execute {
@@ -103,18 +106,17 @@ class RoomListViewModel @Inject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.rx()
+        session.flow()
                 .liveRoomChangeMembershipState()
-                .subscribe {
-                    setState { copy(roomMembershipChanges = it) }
+                .setOnEach {
+                    copy(roomMembershipChanges = it)
                 }
-                .disposeOnClear()
     }
 
-    companion object : MvRxViewModelFactory<RoomListViewModel, RoomListViewState> {
+    companion object : MavericksViewModelFactory<RoomListViewModel, RoomListViewState> {
 
         @JvmStatic
-        override fun create(viewModelContext: ViewModelContext, state: RoomListViewState): RoomListViewModel? {
+        override fun create(viewModelContext: ViewModelContext, state: RoomListViewState): RoomListViewModel {
             val fragment: RoomListFragment = (viewModelContext as FragmentViewModelContext).fragment()
             return fragment.roomListViewModelFactory.create(state)
         }
@@ -320,7 +322,7 @@ class RoomListViewModel @Inject constructor(
 
     private fun String.otherTag(): String? {
         return when (this) {
-            RoomTag.ROOM_TAG_FAVOURITE -> RoomTag.ROOM_TAG_LOW_PRIORITY
+            RoomTag.ROOM_TAG_FAVOURITE    -> RoomTag.ROOM_TAG_LOW_PRIORITY
             RoomTag.ROOM_TAG_LOW_PRIORITY -> RoomTag.ROOM_TAG_FAVOURITE
             else                          -> null
         }

vector/src/main/java/im/vector/app/features/invite/InviteUsersToRoomViewModel.kt

@@ -18,16 +18,23 @@ package im.vector.app.features.invite
 
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
-import com.airbnb.mvrx.MvRxViewModelFactory
+import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
 import dagger.assisted.Assisted
-import dagger.assisted.AssistedInject
 import dagger.assisted.AssistedFactory
+import dagger.assisted.AssistedInject
 import im.vector.app.R
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
 import im.vector.app.features.userdirectory.PendingSelection
-import io.reactivex.Observable
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onCompletion
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.rx.rx
 
@@ -44,7 +51,7 @@ class InviteUsersToRoomViewModel @AssistedInject constructor(@Assisted
         fun create(initialState: InviteUsersToRoomViewState): InviteUsersToRoomViewModel
     }
 
-    companion object : MvRxViewModelFactory<InviteUsersToRoomViewModel, InviteUsersToRoomViewState> {
+    companion object : MavericksViewModelFactory<InviteUsersToRoomViewModel, InviteUsersToRoomViewState> {
 
         @JvmStatic
         override fun create(viewModelContext: ViewModelContext, state: InviteUsersToRoomViewState): InviteUsersToRoomViewModel? {
@@ -63,32 +70,33 @@ class InviteUsersToRoomViewModel @AssistedInject constructor(@Assisted
     }
 
     private fun inviteUsersToRoom(selections: Set<PendingSelection>) {
-        _viewEvents.post(InviteUsersToRoomViewEvents.Loading)
-
-        Observable.fromIterable(selections).flatMapCompletable { user ->
-            when (user) {
-                is PendingSelection.UserPendingSelection     -> room.rx().invite(user.user.userId, null)
-                is PendingSelection.ThreePidPendingSelection -> room.rx().invite3pid(user.threePid)
-            }
-        }.subscribe(
-                {
-                    val successMessage = when (selections.size) {
-                        1    -> stringProvider.getString(R.string.invitation_sent_to_one_user,
-                                selections.first().getBestName())
-                        2    -> stringProvider.getString(R.string.invitations_sent_to_two_users,
-                                selections.first().getBestName(),
-                                selections.last().getBestName())
-                        else -> stringProvider.getQuantityString(R.plurals.invitations_sent_to_one_and_more_users,
-                                selections.size - 1,
-                                selections.first().getBestName(),
-                                selections.size - 1)
+        viewModelScope.launch {
+            _viewEvents.post(InviteUsersToRoomViewEvents.Loading)
+            selections.asFlow()
+                    .map { user ->
+                        when (user) {
+                            is PendingSelection.UserPendingSelection     -> room.invite(user.user.userId, null)
+                            is PendingSelection.ThreePidPendingSelection -> room.invite3pid(user.threePid)
+                        }
                     }
-                    _viewEvents.post(InviteUsersToRoomViewEvents.Success(successMessage))
-                },
-                {
-                    _viewEvents.post(InviteUsersToRoomViewEvents.Failure(it))
-                })
-                .disposeOnClear()
+                    .catch { cause ->
+                        _viewEvents.post(InviteUsersToRoomViewEvents.Failure(cause))
+                    }
+                    .collect {
+                        val successMessage = when (selections.size) {
+                            1    -> stringProvider.getString(R.string.invitation_sent_to_one_user,
+                                    selections.first().getBestName())
+                            2    -> stringProvider.getString(R.string.invitations_sent_to_two_users,
+                                    selections.first().getBestName(),
+                                    selections.last().getBestName())
+                            else -> stringProvider.getQuantityString(R.plurals.invitations_sent_to_one_and_more_users,
+                                    selections.size - 1,
+                                    selections.first().getBestName(),
+                                    selections.size - 1)
+                        }
+                        _viewEvents.post(InviteUsersToRoomViewEvents.Success(successMessage))
+                    }
+        }
     }
 
     fun getUserIdsOfRoomMembers(): Set<String> {

vector/src/main/java/im/vector/app/features/roomprofile/permissions/RoomPermissionsViewModel.kt

@@ -16,14 +16,13 @@
 
 package im.vector.app.features.roomprofile.permissions
 
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MvRxViewModelFactory
 import com.airbnb.mvrx.Success
 import com.airbnb.mvrx.ViewModelContext
 import dagger.assisted.Assisted
-import dagger.assisted.AssistedInject
 import dagger.assisted.AssistedFactory
+import dagger.assisted.AssistedInject
 import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.powerlevel.PowerLevelsObservableFactory
@@ -33,8 +32,8 @@ import org.matrix.android.sdk.api.session.events.model.EventType
 import org.matrix.android.sdk.api.session.events.model.toContent
 import org.matrix.android.sdk.api.session.room.model.PowerLevelsContent
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
-import org.matrix.android.sdk.rx.rx
-import org.matrix.android.sdk.rx.unwrap
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 
 class RoomPermissionsViewModel @AssistedInject constructor(@Assisted initialState: RoomPermissionsViewState,
                                                            private val session: Session)
@@ -62,7 +61,7 @@ class RoomPermissionsViewModel @AssistedInject constructor(@Assisted initialStat
     }
 
     private fun observeRoomSummary() {
-        room.rx().liveRoomSummary()
+        room.flow().liveRoomSummary()
                 .unwrap()
                 .execute { async ->
                     copy(

vector/src/main/java/im/vector/app/features/settings/VectorSettingsGeneralFragment.kt

@@ -51,15 +51,19 @@ import im.vector.app.databinding.DialogChangePasswordBinding
 import im.vector.app.features.MainActivity
 import im.vector.app.features.MainActivityArgs
 import im.vector.app.features.workers.signout.SignOutUiWorker
-import io.reactivex.android.schedulers.AndroidSchedulers
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.distinctUntilChangedBy
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import org.matrix.android.sdk.api.failure.isInvalidPassword
 import org.matrix.android.sdk.api.session.integrationmanager.IntegrationManagerConfig
 import org.matrix.android.sdk.api.session.integrationmanager.IntegrationManagerService
-import org.matrix.android.sdk.rx.rx
-import org.matrix.android.sdk.rx.unwrap
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 import java.io.File
 import java.util.UUID
 import javax.inject.Inject
@@ -118,29 +122,29 @@ class VectorSettingsGeneralFragment @Inject constructor(
     }
 
     private fun observeUserAvatar() {
-        session.rx()
+        session.flow()
                 .liveUser(session.myUserId)
                 .unwrap()
-                .distinctUntilChanged { user -> user.avatarUrl }
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe { mUserAvatarPreference.refreshAvatar(it) }
-                .disposeOnDestroyView()
+                .distinctUntilChangedBy { user -> user.avatarUrl }
+                .onEach {
+                    mUserAvatarPreference.refreshAvatar(it)
+                }
+                .launchIn(viewLifecycleOwner.lifecycleScope)
     }
 
     private fun observeUserDisplayName() {
-        session.rx()
+        session.flow()
                 .liveUser(session.myUserId)
                 .unwrap()
                 .map { it.displayName ?: "" }
                 .distinctUntilChanged()
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe { displayName ->
+                .onEach { displayName ->
                     mDisplayNamePreference.let {
                         it.summary = displayName
                         it.text = displayName
                     }
                 }
-                .disposeOnDestroyView()
+                .launchIn(viewLifecycleOwner.lifecycleScope)
     }
 
     override fun bindPref() {

vector/src/main/java/im/vector/app/features/settings/devtools/AccountDataViewModel.kt

@@ -16,23 +16,23 @@
 
 package im.vector.app.features.settings.devtools
 
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksState
+import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.MvRxViewModelFactory
 import com.airbnb.mvrx.Uninitialized
 import com.airbnb.mvrx.ViewModelContext
 import dagger.assisted.Assisted
-import dagger.assisted.AssistedInject
 import dagger.assisted.AssistedFactory
+import dagger.assisted.AssistedInject
 import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.platform.EmptyViewEvents
 import im.vector.app.core.platform.VectorViewModel
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.accountdata.UserAccountDataEvent
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 
 data class AccountDataViewState(
         val accountData: Async<List<UserAccountDataEvent>> = Uninitialized
@@ -43,7 +43,7 @@ class AccountDataViewModel @AssistedInject constructor(@Assisted initialState: A
     : VectorViewModel<AccountDataViewState, AccountDataAction, EmptyViewEvents>(initialState) {
 
     init {
-        session.rx().liveUserAccountData(emptySet())
+        session.flow().liveUserAccountData(emptySet())
                 .execute {
                     copy(accountData = it)
                 }
@@ -66,7 +66,7 @@ class AccountDataViewModel @AssistedInject constructor(@Assisted initialState: A
         fun create(initialState: AccountDataViewState): AccountDataViewModel
     }
 
-    companion object : MvRxViewModelFactory<AccountDataViewModel, AccountDataViewState> {
+    companion object : MavericksViewModelFactory<AccountDataViewModel, AccountDataViewState> {
 
         @JvmStatic
         override fun create(viewModelContext: ViewModelContext, state: AccountDataViewState): AccountDataViewModel? {