Mavericks 2: continue replacing Rx

vector/src/main/java/im/vector/app/features/discovery/DiscoverySettingsViewModel.kt

@@ -15,7 +15,6 @@
  */
 package im.vector.app.features.discovery
 
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -25,17 +24,19 @@ import com.airbnb.mvrx.Success
 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.VectorViewModel
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.identity.IdentityServiceError
 import org.matrix.android.sdk.api.session.identity.IdentityServiceListener
 import org.matrix.android.sdk.api.session.identity.SharedState
 import org.matrix.android.sdk.api.session.identity.ThreePid
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 
 class DiscoverySettingsViewModel @AssistedInject constructor(
         @Assisted initialState: DiscoverySettingsState,
@@ -84,12 +85,12 @@ class DiscoverySettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeThreePids() {
-        session.rx()
+        session.flow()
                 .liveThreePIds(true)
-                .subscribe {
+                .onEach {
                     retrieveBinding(it)
                 }
-                .disposeOnClear()
+                .launchIn(viewModelScope)
     }
 
     override fun onCleared() {

vector/src/main/java/im/vector/app/features/home/HomeActivity.kt

@@ -27,6 +27,7 @@ import android.view.MenuItem
 import androidx.core.view.GravityCompat
 import androidx.core.view.isVisible
 import androidx.drawerlayout.widget.DrawerLayout
+import androidx.lifecycle.lifecycleScope
 import com.airbnb.mvrx.Mavericks
 import com.airbnb.mvrx.viewModel
 import com.google.android.material.appbar.MaterialToolbar
@@ -72,6 +73,7 @@ import im.vector.app.features.workers.signout.ServerBackupStatusViewModel
 import im.vector.app.features.workers.signout.ServerBackupStatusViewState
 import im.vector.app.push.fcm.FcmHelper
 import io.reactivex.android.schedulers.AndroidSchedulers
+import kotlinx.coroutines.launch
 import kotlinx.parcelize.Parcelize
 import org.matrix.android.sdk.api.session.initsync.SyncStatusService
 import org.matrix.android.sdk.api.session.permalinks.PermalinkService
@@ -288,26 +290,23 @@ class HomeActivity :
                 }
                 else                                                  -> deepLink
             }
-            permalinkHandler.launch(
+            lifecycleScope.launch {
-                    context = this,
+                val isHandled = permalinkHandler.launch(
+                        context = this@HomeActivity,
                         deepLink = resolvedLink,
-                    navigationInterceptor = this,
+                        navigationInterceptor = this@HomeActivity,
                         buildTask = true
                 )
-                    // .delay(500, TimeUnit.MILLISECONDS)
-                    .observeOn(AndroidSchedulers.mainThread())
-                    .subscribe { isHandled ->
                 if (!isHandled) {
                     val isMatrixToLink = deepLink.startsWith(PermalinkService.MATRIX_TO_URL_BASE)
                             || deepLink.startsWith(MATRIX_TO_CUSTOM_SCHEME_URL_BASE)
-                            MaterialAlertDialogBuilder(this)
+                    MaterialAlertDialogBuilder(this@HomeActivity)
                             .setTitle(R.string.dialog_title_error)
                             .setMessage(if (isMatrixToLink) R.string.permalink_malformed else R.string.universal_link_malformed)
                             .setPositiveButton(R.string.ok, null)
                             .show()
                 }
             }
-                    .disposeOnDestroy()
         }
     }
 

vector/src/main/java/im/vector/app/features/home/HomeActivityViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.home
 
+import androidx.lifecycle.asFlow
 import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.Mavericks
 import com.airbnb.mvrx.MavericksViewModelFactory
@@ -31,6 +32,8 @@ import im.vector.app.features.session.coroutineScope
 import im.vector.app.features.settings.VectorPreferences
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.auth.UIABaseAuth
 import org.matrix.android.sdk.api.auth.UserInteractiveAuthInterceptor
@@ -44,6 +47,7 @@ import org.matrix.android.sdk.api.session.initsync.SyncStatusService
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
 import org.matrix.android.sdk.api.util.toMatrixItem
+import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.model.CryptoDeviceInfo
 import org.matrix.android.sdk.internal.crypto.model.MXUsersDevicesMap
 import org.matrix.android.sdk.internal.util.awaitCallback
@@ -100,9 +104,9 @@ class HomeActivityViewModel @AssistedInject constructor(
                 .crossSigningService().allPrivateKeysKnown()
 
         safeActiveSession
-                .rx()
+                .flow()
                 .liveCrossSigningInfo(safeActiveSession.myUserId)
-                .subscribe {
+                .onEach {
                     val isVerified = it.getOrNull()?.isTrusted() ?: false
                     if (!isVerified && onceTrusted) {
                         // cross signing keys have been reset
@@ -116,15 +120,15 @@ class HomeActivityViewModel @AssistedInject constructor(
                     }
                     onceTrusted = isVerified
                 }
-                .disposeOnClear()
+                .launchIn(viewModelScope)
     }
 
     private fun observeInitialSync() {
         val session = activeSessionHolder.getSafeActiveSession() ?: return
 
         session.getSyncStatusLive()
-                .asObservable()
+                .asFlow()
-                .subscribe { status ->
+                .onEach { status ->
                     when (status) {
                         is SyncStatusService.Status.Progressing -> {
                             // Schedule a check of the bootstrap when the init sync will be finished
@@ -145,7 +149,7 @@ class HomeActivityViewModel @AssistedInject constructor(
                         )
                     }
                 }
-                .disposeOnClear()
+                .launchIn(viewModelScope)
     }
 
     /**

vector/src/main/java/im/vector/app/features/home/HomeDetailViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.home
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -37,6 +38,7 @@ import im.vector.app.features.ui.UiStateRepository
 import io.reactivex.schedulers.Schedulers
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.query.ActiveSpaceFilter
 import org.matrix.android.sdk.api.query.RoomCategoryFilter
@@ -48,7 +50,6 @@ import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
 import org.matrix.android.sdk.api.util.toMatrixItem
 import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.rx.asObservable
-import org.matrix.android.sdk.rx.rx
 import timber.log.Timber
 import java.util.concurrent.TimeUnit
 
@@ -182,26 +183,19 @@ class HomeDetailViewModel @AssistedInject constructor(@Assisted initialState: Ho
     }
 
     private fun observeSyncState() {
-        session.rx()
+        session.flow()
                 .liveSyncState()
-                .subscribe { syncState ->
+                .setOnEach { syncState ->
-                    setState {
                     copy(syncState = syncState)
                 }
-                }
-                .disposeOnClear()
 
         session.getSyncStatusLive()
-                .asObservable()
+                .asFlow()
-                .subscribe {
+                .filterIsInstance<SyncStatusService.Status.IncrementalSyncStatus>()
-                    if (it is SyncStatusService.Status.IncrementalSyncStatus) {
+                .setOnEach {
-                        setState {
                     copy(incrementalSyncStatus = it)
                 }
     }
-                }
-                .disposeOnClear()
-    }
 
     private fun observeRoomGroupingMethod() {
         appStateHandler.selectedRoomGroupingObservable

vector/src/main/java/im/vector/app/features/home/room/breadcrumbs/BreadcrumbsViewModel.kt

@@ -30,6 +30,7 @@ import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
+import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.rx.rx
 
 class BreadcrumbsViewModel @AssistedInject constructor(@Assisted initialState: BreadcrumbsViewState,
@@ -61,12 +62,11 @@ class BreadcrumbsViewModel @AssistedInject constructor(@Assisted initialState: B
     // PRIVATE METHODS *****************************************************************************
 
     private fun observeBreadcrumbs() {
-        session.rx()
+        session.flow()
                 .liveBreadcrumbs(roomSummaryQueryParams {
                     displayName = QueryStringValue.NoCondition
                     memberships = listOf(Membership.JOIN)
                 })
-                .observeOn(Schedulers.computation())
                 .execute { asyncBreadcrumbs ->
                     copy(asyncBreadcrumbs = asyncBreadcrumbs)
                 }

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

@@ -1590,10 +1590,11 @@ class RoomDetailFragment @Inject constructor(
         }
     }
 
-// TimelineEventController.Callback ************************************************************
+    // TimelineEventController.Callback ************************************************************
 
     override fun onUrlClicked(url: String, title: String): Boolean {
-        permalinkHandler
+        viewLifecycleOwner.lifecycleScope.launch {
+            val isManaged = permalinkHandler
                     .launch(requireActivity(), url, object : NavigationInterceptor {
                         override fun navToRoom(roomId: String?, eventId: String?, deepLink: Uri?): Boolean {
                             // Same room?
@@ -1616,10 +1617,7 @@ class RoomDetailFragment @Inject constructor(
                             return true
                         }
                     })
-                .subscribeOn(Schedulers.io())
+            if (!isManaged) {
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe { managed ->
-                    if (!managed) {
                 if (title.isValidUrl() && url.isValidUrl() && URL(title).host != URL(url).host) {
                     MaterialAlertDialogBuilder(requireActivity(), R.style.ThemeOverlay_Vector_MaterialAlertDialog_NegativeDestructive)
                             .setTitle(R.string.external_link_confirmation_title)
@@ -1640,7 +1638,6 @@ class RoomDetailFragment @Inject constructor(
                 }
             }
         }
-                .disposeOnDestroyView()
         // In fact it is always managed
         return true
     }
@@ -1799,6 +1796,7 @@ class RoomDetailFragment @Inject constructor(
     }
 
     override fun onRoomCreateLinkClicked(url: String) {
+        viewLifecycleOwner.lifecycleScope.launchWhenResumed {
             permalinkHandler
                     .launch(requireContext(), url, object : NavigationInterceptor {
                         override fun navToRoom(roomId: String?, eventId: String?, deepLink: Uri?): Boolean {
@@ -1806,8 +1804,7 @@ class RoomDetailFragment @Inject constructor(
                             return false
                         }
                     })
-                .subscribe()
+        }
-                .disposeOnDestroyView()
     }
 
     override fun onReadReceiptsClicked(readReceipts: List<ReadReceiptData>) {

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

@@ -18,10 +18,14 @@ package im.vector.app.features.invite
 
 import im.vector.app.ActiveSessionDataSource
 import im.vector.app.features.session.coroutineScope
-import io.reactivex.Observable
 import io.reactivex.disposables.Disposable
 import kotlinx.coroutines.async
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.debounce
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.sync.Semaphore
 import kotlinx.coroutines.sync.withPermit
 import org.matrix.android.sdk.api.extensions.orFalse
@@ -31,9 +35,8 @@ import org.matrix.android.sdk.api.session.room.Room
 import org.matrix.android.sdk.api.session.room.members.ChangeMembershipState
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
-import java.util.concurrent.TimeUnit
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -50,7 +53,7 @@ class InvitesAcceptor @Inject constructor(
 
     private lateinit var activeSessionDisposable: Disposable
     private val shouldRejectRoomIds = mutableSetOf<String>()
-    private val invitedRoomDisposables = HashMap<String, Disposable>()
+    private val activeSessionIds = mutableSetOf<String>()
     private val semaphore = Semaphore(1)
 
     fun initialize() {
@@ -71,23 +74,26 @@ class InvitesAcceptor @Inject constructor(
         if (!autoAcceptInvites.isEnabled) {
             return
         }
-        if (invitedRoomDisposables.containsKey(session.sessionId)) {
+        if (activeSessionIds.contains(session.sessionId)) {
             return
         }
+        activeSessionIds.add(session.sessionId)
         session.addListener(this)
         val roomQueryParams = roomSummaryQueryParams {
             this.memberships = listOf(Membership.INVITE)
         }
-        val rxSession = session.rx()
+        val flowSession = session.flow()
-        Observable
+        combine(
-                .combineLatest(
+                flowSession.liveRoomSummaries(roomQueryParams),
-                        rxSession.liveRoomSummaries(roomQueryParams),
+                flowSession.liveRoomChangeMembershipState().debounce(1000)
-                        rxSession.liveRoomChangeMembershipState().debounce(1, TimeUnit.SECONDS),
+        ) { invitedRooms, _ -> invitedRooms.map { it.roomId } }
-                        { invitedRooms, _ -> invitedRooms.map { it.roomId } }
-                )
                 .filter { it.isNotEmpty() }
-                .subscribe { invitedRoomIds ->
+                .onEach { invitedRoomIds ->
-                    session.coroutineScope.launch {
+                    joinInvitedRooms(session, invitedRoomIds)
+                }.launchIn(session.coroutineScope)
+    }
+
+    private suspend fun joinInvitedRooms(session: Session, invitedRoomIds: List<String>) = coroutineScope {
         semaphore.withPermit {
             Timber.v("Invited roomIds: $invitedRoomIds")
             for (roomId in invitedRoomIds) {
@@ -95,11 +101,6 @@ class InvitesAcceptor @Inject constructor(
             }
         }
     }
-                }
-                .also {
-                    invitedRoomDisposables[session.sessionId] = it
-                }
-    }
 
     private suspend fun Session.joinRoomSafely(roomId: String) {
         if (shouldRejectRoomIds.contains(roomId)) {
@@ -138,6 +139,6 @@ class InvitesAcceptor @Inject constructor(
 
     override fun onSessionStopped(session: Session) {
         session.removeListener(this)
-        invitedRoomDisposables.remove(session.sessionId)?.dispose()
+        activeSessionIds.remove(session.sessionId)
     }
 }

vector/src/main/java/im/vector/app/features/permalink/PermalinkHandler.kt

@@ -23,10 +23,10 @@ import im.vector.app.core.di.ActiveSessionHolder
 import im.vector.app.core.utils.toast
 import im.vector.app.features.navigation.Navigator
 import im.vector.app.features.roomdirectory.roompreview.RoomPreviewData
-import io.reactivex.Single
+import kotlinx.coroutines.Dispatchers
-import io.reactivex.android.schedulers.AndroidSchedulers
+import kotlinx.coroutines.withContext
-import io.reactivex.schedulers.Schedulers
 import org.matrix.android.sdk.api.extensions.orFalse
+import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.session.permalinks.PermalinkData
 import org.matrix.android.sdk.api.session.permalinks.PermalinkParser
 import org.matrix.android.sdk.api.session.permalinks.PermalinkService
@@ -34,56 +34,48 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.RoomType
 import org.matrix.android.sdk.api.util.Optional
 import org.matrix.android.sdk.api.util.toOptional
-import org.matrix.android.sdk.rx.rx
 import javax.inject.Inject
 
 class PermalinkHandler @Inject constructor(private val activeSessionHolder: ActiveSessionHolder,
                                            private val navigator: Navigator) {
 
-    fun launch(
+    suspend fun launch(
             context: Context,
             deepLink: String?,
             navigationInterceptor: NavigationInterceptor? = null,
             buildTask: Boolean = false
-    ): Single<Boolean> {
+    ): Boolean {
         val uri = deepLink?.let { Uri.parse(it) }
         return launch(context, uri, navigationInterceptor, buildTask)
     }
 
-    fun launch(
+    suspend fun launch(
             context: Context,
             deepLink: Uri?,
             navigationInterceptor: NavigationInterceptor? = null,
             buildTask: Boolean = false
-    ): Single<Boolean> {
+    ): Boolean {
         if (deepLink == null || !isPermalinkSupported(context, deepLink.toString())) {
-            return Single.just(false)
+            return false
         }
-        return Single
+        return tryOrNull {
-                .fromCallable {
+            withContext(Dispatchers.Default) {
-                    PermalinkParser.parse(deepLink)
+                val permalinkData = PermalinkParser.parse(deepLink)
-                }
-                .subscribeOn(Schedulers.computation())
-                .observeOn(AndroidSchedulers.mainThread())
-                .flatMap { permalinkData ->
                 handlePermalink(permalinkData, deepLink, context, navigationInterceptor, buildTask)
             }
-                .onErrorReturnItem(false)
+        } ?: false
     }
 
-    private fun handlePermalink(
+    private suspend fun handlePermalink(
             permalinkData: PermalinkData,
             rawLink: Uri,
             context: Context,
             navigationInterceptor: NavigationInterceptor?,
             buildTask: Boolean
-    ): Single<Boolean> {
+    ): Boolean {
         return when (permalinkData) {
             is PermalinkData.RoomLink            -> {
-                permalinkData.getRoomId()
+                val roomId = permalinkData.getRoomId()
-                        .observeOn(AndroidSchedulers.mainThread())
-                        .map {
-                            val roomId = it.getOrNull()
                 if (navigationInterceptor?.navToRoom(roomId, permalinkData.eventId, rawLink) != true) {
                     openRoom(
                             context = context,
@@ -95,19 +87,18 @@ class PermalinkHandler @Inject constructor(private val activeSessionHolder: Acti
                 }
                 true
             }
-            }
             is PermalinkData.GroupLink           -> {
                 navigator.openGroupDetail(permalinkData.groupId, context, buildTask)
-                Single.just(true)
+                true
             }
             is PermalinkData.UserLink            -> {
                 if (navigationInterceptor?.navToMemberProfile(permalinkData.userId, rawLink) != true) {
                     navigator.openRoomMemberProfile(userId = permalinkData.userId, roomId = null, context = context, buildTask = buildTask)
                 }
-                Single.just(true)
+                true
             }
             is PermalinkData.FallbackLink        -> {
-                Single.just(false)
+                false
             }
             is PermalinkData.RoomEmailInviteLink -> {
                 val data = RoomPreviewData(
@@ -118,7 +109,7 @@ class PermalinkHandler @Inject constructor(private val activeSessionHolder: Acti
                         roomType = permalinkData.roomType
                 )
                 navigator.openRoomPreview(context, data)
-                Single.just(true)
+                true
             }
         }
     }
@@ -130,15 +121,13 @@ class PermalinkHandler @Inject constructor(private val activeSessionHolder: Acti
         }
     }
 
-    private fun PermalinkData.RoomLink.getRoomId(): Single<Optional<String>> {
+    private suspend fun PermalinkData.RoomLink.getRoomId(): String? {
         val session = activeSessionHolder.getSafeActiveSession()
         return if (isRoomAlias && session != null) {
-            session.rx()
+            val roomIdByAlias = session.getRoomIdByAlias(roomIdOrAlias, true)
-                    .getRoomIdByAlias(roomIdOrAlias, true)
+            roomIdByAlias.getOrNull()?.roomId
-                    .map { it.getOrNull()?.roomId.toOptional() }
-                    .subscribeOn(Schedulers.io())
         } else {
-            Single.just(Optional.from(roomIdOrAlias))
+            roomIdOrAlias
         }
     }
 

vector/src/main/java/im/vector/app/features/room/RequireActiveMembershipViewModel.kt

@@ -20,16 +20,26 @@ import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
-import com.jakewharton.rxrelay2.BehaviorRelay
 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.extensions.exhaustive
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
 import io.reactivex.Observable
-import io.reactivex.schedulers.Schedulers
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.subscribe
+import kotlinx.coroutines.flow.switchMap
 import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.events.model.EventType
@@ -37,8 +47,8 @@ import org.matrix.android.sdk.api.session.room.Room
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.RoomSummary
 import org.matrix.android.sdk.api.util.Optional
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.rx.unwrap
+import org.matrix.android.sdk.flow.unwrap
 
 /**
  * This ViewModel observe a room summary and notify when the room is left
@@ -66,28 +76,31 @@ class RequireActiveMembershipViewModel @AssistedInject constructor(
         }
     }
 
-    private val roomIdObservable = BehaviorRelay.createDefault(Optional.from(initialState.roomId))
+    private val roomIdFlow = MutableStateFlow(Optional.from(initialState.roomId))
 
     init {
         observeRoomSummary()
     }
 
     private fun observeRoomSummary() {
-        roomIdObservable
+        roomIdFlow
                 .unwrap()
-                .switchMap { roomId ->
+                .flatMapLatest { roomId ->
-                    val room = session.getRoom(roomId) ?: return@switchMap Observable.just(Optional.empty<RequireActiveMembershipViewEvents.RoomLeft>())
+                    val room = session.getRoom(roomId) ?: return@flatMapLatest flow{
-                    room.rx()
+                        val emptyResult = Optional.empty<RequireActiveMembershipViewEvents.RoomLeft>()
+                        emit(emptyResult)
+                    }
+                    room.flow()
                             .liveRoomSummary()
                             .unwrap()
-                            .observeOn(Schedulers.computation())
+                            .flowOn(Dispatchers.Default)
                             .map { mapToLeftViewEvent(room, it) }
                 }
                 .unwrap()
-                .subscribe { event ->
+                .onEach { event ->
                     _viewEvents.post(event)
                 }
-                .disposeOnClear()
+                .launchIn(viewModelScope)
     }
 
     private fun mapToLeftViewEvent(room: Room, roomSummary: RoomSummary): Optional<RequireActiveMembershipViewEvents.RoomLeft> {
@@ -128,7 +141,7 @@ class RequireActiveMembershipViewModel @AssistedInject constructor(
                 setState {
                     copy(roomId = action.roomId)
                 }
-                roomIdObservable.accept(Optional.from(action.roomId))
+                roomIdFlow.tryEmit(Optional.from(action.roomId))
             }
         }.exhaustive
     }

vector/src/main/java/im/vector/app/features/roomdirectory/PublicRoomsFragment.kt

@@ -22,6 +22,7 @@ import android.view.LayoutInflater
 import android.view.MenuItem
 import android.view.View
 import android.view.ViewGroup
+import androidx.lifecycle.lifecycleScope
 import com.airbnb.mvrx.activityViewModel
 import com.airbnb.mvrx.withState
 import com.jakewharton.rxbinding3.appcompat.queryTextChanges
@@ -37,6 +38,7 @@ import im.vector.app.databinding.FragmentPublicRoomsBinding
 import im.vector.app.features.permalink.NavigationInterceptor
 import im.vector.app.features.permalink.PermalinkHandler
 import io.reactivex.rxkotlin.subscribeBy
+import kotlinx.coroutines.launch
 
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.room.model.roomdirectory.PublicRoom
@@ -125,20 +127,20 @@ class PublicRoomsFragment @Inject constructor(
     }
 
     override fun onUnknownRoomClicked(roomIdOrAlias: String) {
+        viewLifecycleOwner.lifecycleScope.launch {
             val permalink = session.permalinkService().createPermalink(roomIdOrAlias)
-        permalinkHandler
+            val isHandled = permalinkHandler
                     .launch(requireContext(), permalink, object : NavigationInterceptor {
                         override fun navToRoom(roomId: String?, eventId: String?, deepLink: Uri?): Boolean {
                             requireActivity().finish()
                             return false
                         }
                     })
-                .subscribe { isSuccessful ->
+
-                    if (!isSuccessful) {
+            if (!isHandled) {
                 requireContext().toast(R.string.room_error_not_found)
             }
         }
-                .disposeOnDestroyView()
     }
 
     override fun onPublicRoomClicked(publicRoom: PublicRoom, joinState: JoinState) {

vector/src/main/java/im/vector/app/features/roomdirectory/RoomDirectoryViewModel.kt

@@ -16,7 +16,6 @@
 
 package im.vector.app.features.roomdirectory
 
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.Loading
@@ -31,6 +30,7 @@ import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.settings.VectorPreferences
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.orFalse
 import org.matrix.android.sdk.api.session.Session
@@ -38,7 +38,7 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.roomdirectory.PublicRoomsFilter
 import org.matrix.android.sdk.api.session.room.model.roomdirectory.PublicRoomsParams
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 
 class RoomDirectoryViewModel @AssistedInject constructor(
@@ -80,28 +80,24 @@ class RoomDirectoryViewModel @AssistedInject constructor(
             memberships = listOf(Membership.JOIN)
         }
         session
-                .rx()
+                .flow()
                 .liveRoomSummaries(queryParams)
-                .subscribe { list ->
+                .map { roomSummaries ->
-                    val joinedRoomIds = list
+                    roomSummaries
-                            ?.map { it.roomId }
+                            .map { it.roomId }
-                            ?.toSet()
+                            .toSet()
-                            .orEmpty()
-
-                    setState {
-                        copy(joinedRoomsIds = joinedRoomIds)
                 }
+                .setOnEach {
+                    copy(joinedRoomsIds = it)
                 }
-                .disposeOnClear()
     }
 
     private fun observeMembershipChanges() {
-        session.rx()
+        session.flow()
                 .liveRoomChangeMembershipState()
-                .subscribe {
+                .setOnEach {
-                    setState { copy(changeMembershipStates = it) }
+                    copy(changeMembershipStates = it)
                 }
-                .disposeOnClear()
     }
 
     override fun handle(action: RoomDirectoryAction) {

vector/src/main/java/im/vector/app/features/roomdirectory/roompreview/RoomPreviewViewModel.kt

@@ -30,6 +30,8 @@ import im.vector.app.core.platform.EmptyViewEvents
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.roomdirectory.JoinState
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.query.QueryStringValue
@@ -40,6 +42,7 @@ import org.matrix.android.sdk.api.session.room.members.ChangeMembershipState
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.peeking.PeekResult
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
+import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.rx.rx
 import timber.log.Timber
 
@@ -165,9 +168,9 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
             excludeType = null
         }
         session
-                .rx()
+                .flow()
                 .liveRoomSummaries(queryParams)
-                .subscribe { list ->
+                .onEach { list ->
                     val isRoomJoined = list.any {
                         it.membership == Membership.JOIN
                     }
@@ -180,13 +183,13 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
                         setState { copy(roomJoinState = JoinState.JOINED) }
                     }
                 }
-                .disposeOnClear()
+                .launchIn(viewModelScope)
     }
 
     private fun observeMembershipChanges() {
-        session.rx()
+        session.flow()
                 .liveRoomChangeMembershipState()
-                .subscribe {
+                .onEach {
                     val changeMembership = it[initialState.roomId] ?: ChangeMembershipState.Unknown
                     val joinState = when (changeMembership) {
                         is ChangeMembershipState.Joining       -> JoinState.JOINING
@@ -198,7 +201,7 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
                         setState { copy(roomJoinState = joinState) }
                     }
                 }
-                .disposeOnClear()
+                .launchIn(viewModelScope)
     }
 
     override fun handle(action: RoomPreviewAction) {

vector/src/main/java/im/vector/app/features/roomprofile/RoomProfileViewModel.kt

@@ -40,10 +40,10 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.create.RoomCreateContent
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.api.session.room.state.isPublic
-import org.matrix.android.sdk.rx.RxRoom
+import org.matrix.android.sdk.flow.FlowRoom
-import org.matrix.android.sdk.rx.mapOptional
+import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.mapOptional
-import org.matrix.android.sdk.rx.unwrap
+import org.matrix.android.sdk.flow.unwrap
 
 class RoomProfileViewModel @AssistedInject constructor(
         @Assisted private val initialState: RoomProfileViewState,
@@ -69,15 +69,15 @@ class RoomProfileViewModel @AssistedInject constructor(
     private val room = session.getRoom(initialState.roomId)!!
 
     init {
-        val rxRoom = room.rx()
+        val flowRoom = room.flow()
-        observeRoomSummary(rxRoom)
+        observeRoomSummary(flowRoom)
-        observeRoomCreateContent(rxRoom)
+        observeRoomCreateContent(flowRoom)
-        observeBannedRoomMembers(rxRoom)
+        observeBannedRoomMembers(flowRoom)
         observePermissions()
     }
 
-    private fun observeRoomCreateContent(rxRoom: RxRoom) {
+    private fun observeRoomCreateContent(flowRoom: FlowRoom) {
-        rxRoom.liveStateEvent(EventType.STATE_ROOM_CREATE, QueryStringValue.NoCondition)
+        flowRoom.liveStateEvent(EventType.STATE_ROOM_CREATE, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomCreateContent>() }
                 .unwrap()
                 .execute { async ->
@@ -92,16 +92,16 @@ class RoomProfileViewModel @AssistedInject constructor(
                 }
     }
 
-    private fun observeRoomSummary(rxRoom: RxRoom) {
+    private fun observeRoomSummary(flowRoom: FlowRoom) {
-        rxRoom.liveRoomSummary()
+        flowRoom.liveRoomSummary()
                 .unwrap()
                 .execute {
                     copy(roomSummary = it)
                 }
     }
 
-    private fun observeBannedRoomMembers(rxRoom: RxRoom) {
+    private fun observeBannedRoomMembers(flowRoom: FlowRoom) {
-        rxRoom.liveRoomMembers(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
+        flowRoom.liveRoomMembers(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
                 .execute {
                     copy(bannedMembership = it)
                 }

vector/src/main/java/im/vector/app/features/roomprofile/settings/RoomSettingsViewModel.kt

@@ -28,11 +28,10 @@ import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import im.vector.app.features.settings.VectorPreferences
-import io.reactivex.Completable
-import io.reactivex.Observable
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.Session
@@ -47,7 +46,6 @@ import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.flow.mapOptional
 import org.matrix.android.sdk.flow.unwrap
-import org.matrix.android.sdk.rx.rx
 
 class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState: RoomSettingsViewState,
                                                         private val vectorPreferences: VectorPreferences,
@@ -259,42 +257,39 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun saveSettings() = withState { state ->
-        postLoading(true)
+        val operationList = mutableListOf<suspend () -> Unit>()
-
-        val operationList = mutableListOf<Completable>()
 
         val summary = state.roomSummary.invoke()
 
         when (val avatarAction = state.avatarAction) {
             RoomSettingsViewState.AvatarAction.None            -> Unit
             RoomSettingsViewState.AvatarAction.DeleteAvatar    -> {
-                operationList.add(room.rx().deleteAvatar())
+                operationList.add { room.deleteAvatar() }
             }
             is RoomSettingsViewState.AvatarAction.UpdateAvatar -> {
-                operationList.add(room.rx().updateAvatar(avatarAction.newAvatarUri, avatarAction.newAvatarFileName))
+                operationList.add { room.updateAvatar(avatarAction.newAvatarUri, avatarAction.newAvatarFileName) }
             }
         }
         if (summary?.name != state.newName) {
-            operationList.add(room.rx().updateName(state.newName ?: ""))
+            operationList.add { room.updateName(state.newName ?: "") }
         }
         if (summary?.topic != state.newTopic) {
-            operationList.add(room.rx().updateTopic(state.newTopic ?: ""))
+            operationList.add { room.updateTopic(state.newTopic ?: "") }
         }
 
         if (state.newHistoryVisibility != null) {
-            operationList.add(room.rx().updateHistoryReadability(state.newHistoryVisibility))
+            operationList.add { room.updateHistoryReadability(state.newHistoryVisibility) }
         }
 
         if (state.newRoomJoinRules.hasChanged()) {
-            operationList.add(room.rx().updateJoinRule(state.newRoomJoinRules.newJoinRules, state.newRoomJoinRules.newGuestAccess))
+            operationList.add { room.updateJoinRule(state.newRoomJoinRules.newJoinRules, state.newRoomJoinRules.newGuestAccess) }
+        }
+        viewModelScope.launch {
+            updateLoadingState(isLoading = true)
+            try {
+                for (operation in operationList) {
+                    operation.invoke()
                 }
-
-        Observable
-                .fromIterable(operationList)
-                .concatMapCompletable { it }
-                .subscribe(
-                        {
-                            postLoading(false)
                 setState {
                     deletePendingAvatar(this)
                     copy(
@@ -304,16 +299,15 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
                     )
                 }
                 _viewEvents.post(RoomSettingsViewEvents.Success)
-                        },
+            } catch (failure: Throwable) {
-                        {
+                _viewEvents.post(RoomSettingsViewEvents.Failure(failure))
-                            postLoading(false)
+            }finally {
-                            _viewEvents.post(RoomSettingsViewEvents.Failure(it))
+                updateLoadingState(isLoading = false)
+            }
         }
-                )
-                .disposeOnClear()
     }
 
-    private fun postLoading(isLoading: Boolean) {
+    private fun updateLoadingState(isLoading: Boolean) {
         setState {
             copy(isLoading = isLoading)
         }

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

@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) 2021 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package im.vector.app.features.settings
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.distinctUntilChanged
+import org.matrix.android.sdk.api.extensions.orFalse
+import org.matrix.android.sdk.api.session.Session
+import org.matrix.android.sdk.api.session.crypto.crosssigning.KEYBACKUP_SECRET_SSSS_NAME
+import org.matrix.android.sdk.api.session.crypto.crosssigning.MASTER_KEY_SSSS_NAME
+import org.matrix.android.sdk.api.session.crypto.crosssigning.SELF_SIGNING_KEY_SSSS_NAME
+import org.matrix.android.sdk.api.session.crypto.crosssigning.USER_SIGNING_KEY_SSSS_NAME
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.rx.SecretsSynchronisationInfo
+
+data class SecretsSynchronisationInfo(
+        val isBackupSetup: Boolean,
+        val isCrossSigningEnabled: Boolean,
+        val isCrossSigningTrusted: Boolean,
+        val allPrivateKeysKnown: Boolean,
+        val megolmBackupAvailable: Boolean,
+        val megolmSecretKnown: Boolean,
+        val isMegolmKeyIn4S: Boolean
+)
+
+fun Session.liveSecretSynchronisationInfo(): Flow<SecretsSynchronisationInfo> {
+    val sessionFlow = flow()
+    return combine(
+            sessionFlow.liveUserAccountData(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME, KEYBACKUP_SECRET_SSSS_NAME)),
+            sessionFlow.liveCrossSigningInfo(myUserId),
+            sessionFlow.liveCrossSigningPrivateKeys()
+    ) { _, crossSigningInfo, pInfo ->
+        // first check if 4S is already setup
+        val is4SSetup = sharedSecretStorageService.isRecoverySetup()
+        val isCrossSigningEnabled = crossSigningInfo.getOrNull() != null
+        val isCrossSigningTrusted = crossSigningInfo.getOrNull()?.isTrusted() == true
+        val allPrivateKeysKnown = pInfo.getOrNull()?.allKnown().orFalse()
+
+        val keysBackupService = cryptoService().keysBackupService()
+        val currentBackupVersion = keysBackupService.currentBackupVersion
+        val megolmBackupAvailable = currentBackupVersion != null
+        val savedBackupKey = keysBackupService.getKeyBackupRecoveryKeyInfo()
+
+        val megolmKeyKnown = savedBackupKey?.version == currentBackupVersion
+        SecretsSynchronisationInfo(
+                isBackupSetup = is4SSetup,
+                isCrossSigningEnabled = isCrossSigningEnabled,
+                isCrossSigningTrusted = isCrossSigningTrusted,
+                allPrivateKeysKnown = allPrivateKeysKnown,
+                megolmBackupAvailable = megolmBackupAvailable,
+                megolmSecretKnown = megolmKeyKnown,
+                isMegolmKeyIn4S = sharedSecretStorageService.isMegolmKeyInBackup()
+        )
+    }
+            .distinctUntilChanged()
+}

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

@@ -60,6 +60,10 @@ import im.vector.app.features.raw.wellknown.isE2EByDefault
 import im.vector.app.features.themes.ThemeUtils
 import io.reactivex.android.schedulers.AndroidSchedulers
 import io.reactivex.disposables.Disposable
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import me.gujun.android.span.span
 import org.matrix.android.sdk.api.MatrixCallback
@@ -144,14 +148,12 @@ class VectorSettingsSecurityPrivacyFragment @Inject constructor(
         // My device name may have been updated
         refreshMyDevice()
         refreshXSigningStatus()
-        session.rx().liveSecretSynchronisationInfo()
+        session.liveSecretSynchronisationInfo()
-                .observeOn(AndroidSchedulers.mainThread())
+                .flowOn(Dispatchers.Main)
-                .subscribe {
+                .onEach {
                     refresh4SSection(it)
                     refreshXSigningStatus()
-                }.also {
+                }.launchIn(viewLifecycleOwner.lifecycleScope)
-                    disposables.add(it)
-                }
 
         lifecycleScope.launchWhenResumed {
             findPreference<VectorPreference>(VectorPreferences.SETTINGS_CRYPTOGRAPHY_HS_ADMIN_DISABLED_E2E_DEFAULT)?.isVisible =

vector/src/main/java/im/vector/app/features/settings/ignored/IgnoredUsersViewModel.kt

@@ -16,7 +16,6 @@
 
 package im.vector.app.features.settings.ignored
 
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -27,14 +26,14 @@ import com.airbnb.mvrx.Success
 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.platform.VectorViewModel
 import im.vector.app.core.platform.VectorViewModelAction
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.user.model.User
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 
 data class IgnoredUsersViewState(
         val ignoredUsers: List<User> = emptyList(),
@@ -68,7 +67,7 @@ class IgnoredUsersViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeIgnoredUsers() {
-        session.rx()
+        session.flow()
                 .liveIgnoredUsers()
                 .execute { async ->
                     copy(

vector/src/main/java/im/vector/app/features/share/IncomingShareViewModel.kt

@@ -19,24 +19,25 @@ package im.vector.app.features.share
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
-import com.jakewharton.rxrelay2.BehaviorRelay
 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.extensions.toggle
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.attachments.isPreviewable
 import im.vector.app.features.attachments.toGroupedContentAttachmentData
 import im.vector.app.features.home.room.list.BreadcrumbsRoomComparator
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.sample
 import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.content.ContentAttachmentData
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
 import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.rx.rx
-import java.util.concurrent.TimeUnit
 
 class IncomingShareViewModel @AssistedInject constructor(
         @Assisted initialState: IncomingShareViewState,
@@ -58,7 +59,7 @@ class IncomingShareViewModel @AssistedInject constructor(
         }
     }
 
-    private val filterStream: BehaviorRelay<String> = BehaviorRelay.createDefault("")
+    private val filterStream = MutableStateFlow("")
 
     init {
         observeRoomSummaries()
@@ -75,7 +76,7 @@ class IncomingShareViewModel @AssistedInject constructor(
                 }
 
         filterStream
-                .switchMap { filter ->
+                .flatMapLatest { filter ->
                     val displayNameQuery = if (filter.isEmpty()) {
                         QueryStringValue.NoCondition
                     } else {
@@ -85,9 +86,9 @@ class IncomingShareViewModel @AssistedInject constructor(
                         displayName = displayNameQuery
                         memberships = listOf(Membership.JOIN)
                     }
-                    session.rx().liveRoomSummaries(filterQueryParams)
+                    session.flow().liveRoomSummaries(filterQueryParams)
                 }
-                .throttleLast(300, TimeUnit.MILLISECONDS)
+                .sample(300)
                 .map { it.sortedWith(breadcrumbsRoomComparator) }
                 .execute {
                     copy(filteredRoomSummaries = it)
@@ -110,7 +111,7 @@ class IncomingShareViewModel @AssistedInject constructor(
     }
 
     private fun handleFilter(action: IncomingShareAction.FilterWith) {
-        filterStream.accept(action.filter)
+        filterStream.tryEmit(action.filter)
     }
 
     private fun handleShareToSelectedRooms() = withState { state ->

vector/src/main/java/im/vector/app/features/spaces/SpacesListViewModel.kt

@@ -16,7 +16,7 @@
 
 package im.vector.app.features.spaces
 
-import androidx.lifecycle.viewModelScope
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
 import com.airbnb.mvrx.MavericksViewModelFactory
@@ -33,8 +33,9 @@ import im.vector.app.features.session.coroutineScope
 import im.vector.app.features.settings.VectorPreferences
 import im.vector.app.group
 import im.vector.app.space
-import io.reactivex.Observable
 import io.reactivex.schedulers.Schedulers
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.query.ActiveSpaceFilter
@@ -44,19 +45,16 @@ import org.matrix.android.sdk.api.session.events.model.toContent
 import org.matrix.android.sdk.api.session.events.model.toModel
 import org.matrix.android.sdk.api.session.group.groupSummaryQueryParams
 import org.matrix.android.sdk.api.session.room.RoomSortOrder
-import org.matrix.android.sdk.api.session.room.accountdata.RoomAccountDataEvent
 import org.matrix.android.sdk.api.session.room.accountdata.RoomAccountDataTypes
 import org.matrix.android.sdk.api.session.room.model.Membership
-import org.matrix.android.sdk.api.session.room.model.RoomSummary
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
 import org.matrix.android.sdk.api.session.room.summary.RoomAggregateNotificationCount
 import org.matrix.android.sdk.api.session.space.SpaceOrderUtils
 import org.matrix.android.sdk.api.session.space.model.SpaceOrderContent
 import org.matrix.android.sdk.api.session.space.model.TopLevelSpaceComparator
-import org.matrix.android.sdk.api.session.user.model.User
 import org.matrix.android.sdk.api.util.toMatrixItem
+import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.rx.asObservable
-import org.matrix.android.sdk.rx.rx
 import java.util.concurrent.TimeUnit
 
 class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: SpaceListViewState,
@@ -286,21 +284,23 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
                     null)
         }
 
-        val rxSession = session.rx()
+        val flowSession = session.flow()
 
-        Observable.combineLatest<User?, List<RoomSummary>, List<RoomAccountDataEvent>, List<RoomSummary>>(
+        combine(
-                rxSession
+                flowSession
                         .liveUser(session.myUserId)
                         .map {
                             it.getOrNull()
                         },
-                rxSession
+                flowSession
                         .liveSpaceSummaries(spaceSummaryQueryParams),
-                session.accountDataService().getLiveRoomAccountDataEvents(setOf(RoomAccountDataTypes.EVENT_TYPE_SPACE_ORDER)).asObservable(),
+                session
-                { _, communityGroups, _ ->
+                        .accountDataService()
+                        .getLiveRoomAccountDataEvents(setOf(RoomAccountDataTypes.EVENT_TYPE_SPACE_ORDER))
+                        .asFlow()
+        ) { _, communityGroups, _ ->
             communityGroups
         }
-        )
                 .execute { async ->
                     val rootSpaces = session.spaceService().getRootSpaceSummaries()
                     val orders = rootSpaces.map {

vector/src/main/java/im/vector/app/features/spaces/explore/SpaceDirectoryFragment.kt

@@ -25,6 +25,7 @@ import android.view.View
 import android.view.ViewGroup
 import androidx.core.text.toSpannable
 import androidx.core.view.isVisible
+import androidx.lifecycle.lifecycleScope
 import com.airbnb.epoxy.EpoxyVisibilityTracker
 import com.airbnb.mvrx.activityViewModel
 import com.airbnb.mvrx.withState
@@ -46,8 +47,7 @@ import im.vector.app.features.permalink.PermalinkHandler
 import im.vector.app.features.spaces.manage.ManageType
 import im.vector.app.features.spaces.manage.SpaceAddRoomSpaceChooserBottomSheet
 import im.vector.app.features.spaces.manage.SpaceManageActivity
-import io.reactivex.android.schedulers.AndroidSchedulers
+import kotlinx.coroutines.launch
-import io.reactivex.schedulers.Schedulers
 import kotlinx.parcelize.Parcelize
 import org.matrix.android.sdk.api.session.room.model.SpaceChildInfo
 import java.net.URL
@@ -200,12 +200,9 @@ class SpaceDirectoryFragment @Inject constructor(
     }
 
     override fun onUrlClicked(url: String, title: String): Boolean {
-        permalinkHandler
+        viewLifecycleOwner.lifecycleScope.launch {
-                .launch(requireActivity(), url, null)
+            val isHandled = permalinkHandler.launch(requireActivity(), url, null)
-                .subscribeOn(Schedulers.io())
+            if (!isHandled) {
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe { managed ->
-                    if (!managed) {
                 if (title.isValidUrl() && url.isValidUrl() && URL(title).host != URL(url).host) {
                     MaterialAlertDialogBuilder(requireActivity(), R.style.ThemeOverlay_Vector_MaterialAlertDialog_Destructive)
                             .setTitle(R.string.external_link_confirmation_title)
@@ -226,7 +223,6 @@ class SpaceDirectoryFragment @Inject constructor(
                 }
             }
         }
-                .disposeOnDestroyView()
         // In fact it is always managed
         return true
     }

vector/src/main/java/im/vector/app/features/spaces/explore/SpaceDirectoryViewModel.kt

@@ -31,6 +31,7 @@ import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
@@ -42,7 +43,7 @@ import org.matrix.android.sdk.api.session.room.model.RoomType
 import org.matrix.android.sdk.api.session.room.model.SpaceChildInfo
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 
 class SpaceDirectoryViewModel @AssistedInject constructor(
@@ -147,7 +148,7 @@ class SpaceDirectoryViewModel @AssistedInject constructor(
             excludeType = null
         }
         session
-                .rx()
+                .flow()
                 .liveRoomSummaries(queryParams)
                 .map {
                     it.map { it.roomId }.toSet()
@@ -158,12 +159,11 @@ class SpaceDirectoryViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.rx()
+        session.flow()
                 .liveRoomChangeMembershipState()
-                .subscribe {
+                .setOnEach {
-                    setState { copy(changeMembershipStates = it) }
+                    copy(changeMembershipStates = it)
                 }
-                .disposeOnClear()
     }
 
     override fun handle(action: SpaceDirectoryViewAction) {

vector/src/main/java/im/vector/app/features/userdirectory/UserListViewModel.kt

@@ -160,16 +160,15 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     }
 
     private fun observeUsers() = withState { state ->
-
         identityServerUsersSearch
                 .filter { it.isEmail() }
                 .throttleLast(300, TimeUnit.MILLISECONDS)
                 .switchMapSingle { search ->
-                    val rx = session.rx()
+                    val flowSession = session.rx()
                     val stream =
-                            rx.lookupThreePid(ThreePid.Email(search)).flatMap {
+                            flowSession.lookupThreePid(ThreePid.Email(search)).flatMap {
                                 it.getOrNull()?.let { foundThreePid ->
-                                    rx.getProfileInfo(foundThreePid.matrixId)
+                                    flowSession.getProfileInfo(foundThreePid.matrixId)
                                             .map { json ->
                                                 ThreePidUser(
                                                         email = search,

vector/src/main/java/im/vector/app/features/widgets/WidgetViewModel.kt

@@ -30,6 +30,7 @@ import dagger.assisted.AssistedInject
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
 import im.vector.app.features.widgets.permissions.WidgetPermissionsHelper
+import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.query.QueryStringValue
@@ -44,7 +45,6 @@ import org.matrix.android.sdk.api.session.widgets.WidgetManagementFailure
 import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.flow.mapOptional
 import org.matrix.android.sdk.flow.unwrap
-import org.matrix.android.sdk.rx.rx
 import timber.log.Timber
 import javax.net.ssl.HttpsURLConnection
 
@@ -135,7 +135,7 @@ class WidgetViewModel @AssistedInject constructor(@Assisted val initialState: Wi
             return
         }
         val widgetId = initialState.widgetId ?: return
-        session.rx()
+        session.flow()
                 .liveRoomWidgets(initialState.roomId, QueryStringValue.Equals(widgetId))
                 .filter { it.isNotEmpty() }
                 .map { it.first() }

vector/src/main/java/im/vector/app/features/widgets/permissions/RoomWidgetPermissionViewModel.kt

@@ -15,23 +15,24 @@
  */
 package im.vector.app.features.widgets.permissions
 
-import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
 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 kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.orFalse
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.widgets.model.WidgetType
-import org.matrix.android.sdk.rx.rx
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 import java.net.URL
 
@@ -48,7 +49,7 @@ class RoomWidgetPermissionViewModel @AssistedInject constructor(@Assisted val in
 
     private fun observeWidget() {
         val widgetId = initialState.widgetId ?: return
-        session.rx()
+        session.flow()
                 .liveRoomWidgets(initialState.roomId, QueryStringValue.Equals(widgetId))
                 .filter { it.isNotEmpty() }
                 .map {