Mavericks 2: remove matrix-sdk-android-flow as it will be easier when entirely migrating to flow

matrix-sdk-android-flow/.gitignore

@@ -1 +0,0 @@
-/build

matrix-sdk-android-flow/build.gradle

@@ -1,49 +0,0 @@
-
-plugins {
-    id 'com.android.library'
-    id 'org.jetbrains.kotlin.android'
-}
-
-android {
-    compileSdk 31
-
-    defaultConfig {
-        minSdk 21
-        targetSdk 31
-
-        testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
-        consumerProguardFiles "consumer-rules.pro"
-    }
-
-    buildTypes {
-        release {
-            minifyEnabled false
-            proguardFiles getDefaultProguardFile('proguard-android-optimize.txt'), 'proguard-rules.pro'
-        }
-    }
-    compileOptions {
-        sourceCompatibility JavaVersion.VERSION_1_8
-        targetCompatibility JavaVersion.VERSION_1_8
-    }
-    kotlinOptions {
-        jvmTarget = '1.8'
-    }
-}
-
-dependencies {
-
-    implementation project(":matrix-sdk-android")
-    implementation libs.androidx.appCompat
-
-    implementation libs.jetbrains.kotlinStdlibJdk7
-    implementation libs.jetbrains.coroutinesCore
-    implementation libs.jetbrains.coroutinesAndroid
-    implementation libs.androidx.lifecycleLivedata
-
-    // Paging
-    implementation libs.androidx.pagingRuntimeKtx
-
-    // Logging
-    implementation libs.jakewharton.timber
-
-}

matrix-sdk-android-flow/proguard-rules.pro

@@ -1,21 +0,0 @@
-# Add project specific ProGuard rules here.
-# You can control the set of applied configuration files using the
-# proguardFiles setting in build.gradle.
-#
-# For more details, see
-#   http://developer.android.com/guide/developing/tools/proguard.html
-
-# If your project uses WebView with JS, uncomment the following
-# and specify the fully qualified class name to the JavaScript interface
-# class:
-#-keepclassmembers class fqcn.of.javascript.interface.for.webview {
-#   public *;
-#}
-
-# Uncomment this to preserve the line number information for
-# debugging stack traces.
-#-keepattributes SourceFile,LineNumberTable
-
-# If you keep the line number information, uncomment this to
-# hide the original source file name.
-#-renamesourcefileattribute SourceFile

matrix-sdk-android-flow/src/androidTest/java/org/matrix/android/sdk/flow/ExampleInstrumentedTest.kt

@@ -1,40 +0,0 @@
-/*
- * 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 org.matrix.android.sdk.flow
-
-import androidx.test.platform.app.InstrumentationRegistry
-import androidx.test.ext.junit.runners.AndroidJUnit4
-
-import org.junit.Test
-import org.junit.runner.RunWith
-
-import org.junit.Assert.*
-
-/**
- * Instrumented test, which will execute on an Android device.
- *
- * See [testing documentation](http://d.android.com/tools/testing).
- */
-@RunWith(AndroidJUnit4::class)
-class ExampleInstrumentedTest {
-    @Test
-    fun useAppContext() {
-        // Context of the app under test.
-        val appContext = InstrumentationRegistry.getInstrumentation().targetContext
-        assertEquals("org.matrix.android.sdk.flow.test", appContext.packageName)
-    }
-}

matrix-sdk-android-flow/src/main/AndroidManifest.xml

@@ -1,5 +0,0 @@
-<?xml version="1.0" encoding="utf-8"?>
-<manifest xmlns:android="http://schemas.android.com/apk/res/android"
-    package="org.matrix.android.sdk.flow">
-
-</manifest>

matrix-sdk-android-flow/src/main/java/org/matrix/android/sdk/flow/FlowRoom.kt

@@ -1,83 +0,0 @@
-/*
- * Copyright 2020 The Matrix.org Foundation C.I.C.
- *
- * 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 org.matrix.android.sdk.flow
-
-import androidx.lifecycle.asFlow
-import kotlinx.coroutines.flow.Flow
-import org.matrix.android.sdk.api.query.QueryStringValue
-import org.matrix.android.sdk.api.session.events.model.Event
-import org.matrix.android.sdk.api.session.room.Room
-import org.matrix.android.sdk.api.session.room.members.RoomMemberQueryParams
-import org.matrix.android.sdk.api.session.room.model.EventAnnotationsSummary
-import org.matrix.android.sdk.api.session.room.model.ReadReceipt
-import org.matrix.android.sdk.api.session.room.model.RoomMemberSummary
-import org.matrix.android.sdk.api.session.room.model.RoomSummary
-import org.matrix.android.sdk.api.session.room.notification.RoomNotificationState
-import org.matrix.android.sdk.api.session.room.send.UserDraft
-import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
-import org.matrix.android.sdk.api.util.Optional
-
-class FlowRoom(private val room: Room) {
-
-    fun liveRoomSummary(): Flow<Optional<RoomSummary>> {
-        return room.getRoomSummaryLive().asFlow()
-    }
-
-    fun liveRoomMembers(queryParams: RoomMemberQueryParams): Flow<List<RoomMemberSummary>> {
-        return room.getRoomMembersLive(queryParams).asFlow()
-    }
-
-    fun liveAnnotationSummary(eventId: String): Flow<Optional<EventAnnotationsSummary>> {
-        return room.getEventAnnotationsSummaryLive(eventId).asFlow()
-    }
-
-    fun liveTimelineEvent(eventId: String): Flow<Optional<TimelineEvent>> {
-        return room.getTimeLineEventLive(eventId).asFlow()
-    }
-
-    fun liveStateEvent(eventType: String, stateKey: QueryStringValue): Flow<Optional<Event>> {
-        return room.getStateEventLive(eventType, stateKey).asFlow()
-    }
-
-    fun liveStateEvents(eventTypes: Set<String>): Flow<List<Event>> {
-        return room.getStateEventsLive(eventTypes).asFlow()
-    }
-
-    fun liveReadMarker(): Flow<Optional<String>> {
-        return room.getReadMarkerLive().asFlow()
-    }
-
-    fun liveReadReceipt(): Flow<Optional<String>> {
-        return room.getMyReadReceiptLive().asFlow()
-    }
-
-    fun liveEventReadReceipts(eventId: String): Flow<List<ReadReceipt>> {
-        return room.getEventReadReceiptsLive(eventId).asFlow()
-    }
-
-    fun liveDraft(): Flow<Optional<UserDraft>> {
-        return room.getDraftLive().asFlow()
-    }
-
-    fun liveNotificationState(): Flow<RoomNotificationState> {
-        return room.getLiveRoomNotificationState().asFlow()
-    }
-}
-
-fun Room.flow(): FlowRoom {
-    return FlowRoom(this)
-}

matrix-sdk-android-flow/src/main/java/org/matrix/android/sdk/flow/FlowSession.kt

@@ -1,138 +0,0 @@
-/*
- * Copyright 2020 The Matrix.org Foundation C.I.C.
- *
- * 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 org.matrix.android.sdk.flow
-
-import androidx.lifecycle.asFlow
-import androidx.paging.PagedList
-import kotlinx.coroutines.flow.Flow
-import org.matrix.android.sdk.api.query.QueryStringValue
-import org.matrix.android.sdk.api.session.Session
-import org.matrix.android.sdk.api.session.accountdata.UserAccountDataEvent
-import org.matrix.android.sdk.api.session.crypto.crosssigning.MXCrossSigningInfo
-import org.matrix.android.sdk.api.session.group.GroupSummaryQueryParams
-import org.matrix.android.sdk.api.session.group.model.GroupSummary
-import org.matrix.android.sdk.api.session.identity.ThreePid
-import org.matrix.android.sdk.api.session.pushers.Pusher
-import org.matrix.android.sdk.api.session.room.RoomSummaryQueryParams
-import org.matrix.android.sdk.api.session.room.accountdata.RoomAccountDataEvent
-import org.matrix.android.sdk.api.session.room.members.ChangeMembershipState
-import org.matrix.android.sdk.api.session.room.model.RoomMemberSummary
-import org.matrix.android.sdk.api.session.room.model.RoomSummary
-import org.matrix.android.sdk.api.session.space.SpaceSummaryQueryParams
-import org.matrix.android.sdk.api.session.sync.SyncState
-import org.matrix.android.sdk.api.session.user.model.User
-import org.matrix.android.sdk.api.session.widgets.model.Widget
-import org.matrix.android.sdk.api.util.Optional
-import org.matrix.android.sdk.internal.crypto.model.CryptoDeviceInfo
-import org.matrix.android.sdk.internal.crypto.model.rest.DeviceInfo
-import org.matrix.android.sdk.internal.crypto.store.PrivateKeysInfo
-
-class RxFlow(private val session: Session) {
-
-    fun liveRoomSummaries(queryParams: RoomSummaryQueryParams): Flow<List<RoomSummary>> {
-        return session.getRoomSummariesLive(queryParams).asFlow()
-    }
-
-    fun liveGroupSummaries(queryParams: GroupSummaryQueryParams): Flow<List<GroupSummary>> {
-        return session.getGroupSummariesLive(queryParams).asFlow()
-    }
-
-    fun liveSpaceSummaries(queryParams: SpaceSummaryQueryParams): Flow<List<RoomSummary>> {
-        return session.spaceService().getSpaceSummariesLive(queryParams).asFlow()
-    }
-
-    fun liveBreadcrumbs(queryParams: RoomSummaryQueryParams): Flow<List<RoomSummary>> {
-        return session.getBreadcrumbsLive(queryParams).asFlow()
-    }
-
-    fun liveMyDevicesInfo(): Flow<List<DeviceInfo>> {
-        return session.cryptoService().getLiveMyDevicesInfo().asFlow()
-    }
-
-    fun liveSyncState(): Flow<SyncState> {
-        return session.getSyncStateLive().asFlow()
-    }
-
-    fun livePushers(): Flow<List<Pusher>> {
-        return session.getPushersLive().asFlow()
-    }
-
-    fun liveUser(userId: String): Flow<Optional<User>> {
-        return session.getUserLive(userId).asFlow()
-    }
-
-    fun liveRoomMember(userId: String, roomId: String): Flow<Optional<RoomMemberSummary>> {
-        return session.getRoomMemberLive(userId, roomId).asFlow()
-    }
-
-    fun liveUsers(): Flow<List<User>> {
-        return session.getUsersLive().asFlow()
-    }
-
-    fun liveIgnoredUsers(): Flow<List<User>> {
-        return session.getIgnoredUsersLive().asFlow()
-    }
-
-    fun livePagedUsers(filter: String? = null, excludedUserIds: Set<String>? = null): Flow<PagedList<User>> {
-        return session.getPagedUsersLive(filter, excludedUserIds).asFlow()
-    }
-
-    fun liveThreePIds(refreshData: Boolean): Flow<List<ThreePid>> {
-        return session.getThreePidsLive(refreshData).asFlow()
-    }
-
-    fun livePendingThreePIds(): Flow<List<ThreePid>> {
-        return session.getPendingThreePidsLive().asFlow()
-    }
-
-    fun liveUserCryptoDevices(userId: String): Flow<List<CryptoDeviceInfo>> {
-        return session.cryptoService().getLiveCryptoDeviceInfo(userId).asFlow()
-    }
-
-    fun liveCrossSigningInfo(userId: String): Flow<Optional<MXCrossSigningInfo>> {
-        return session.cryptoService().crossSigningService().getLiveCrossSigningKeys(userId).asFlow()
-    }
-
-    fun liveCrossSigningPrivateKeys(): Flow<Optional<PrivateKeysInfo>> {
-        return session.cryptoService().crossSigningService().getLiveCrossSigningPrivateKeys().asFlow()
-    }
-
-    fun liveUserAccountData(types: Set<String>): Flow<List<UserAccountDataEvent>> {
-        return session.accountDataService().getLiveUserAccountDataEvents(types).asFlow()
-    }
-
-    fun liveRoomAccountData(types: Set<String>): Flow<List<RoomAccountDataEvent>> {
-        return session.accountDataService().getLiveRoomAccountDataEvents(types).asFlow()
-    }
-
-    fun liveRoomWidgets(
-            roomId: String,
-            widgetId: QueryStringValue,
-            widgetTypes: Set<String>? = null,
-            excludedTypes: Set<String>? = null
-    ): Flow<List<Widget>> {
-        return session.widgetService().getRoomWidgetsLive(roomId, widgetId, widgetTypes, excludedTypes).asFlow()
-    }
-
-    fun liveRoomChangeMembershipState(): Flow<Map<String, ChangeMembershipState>> {
-        return session.getChangeMembershipsLive().asFlow()
-    }
-}
-
-fun Session.flow(): RxFlow {
-    return RxFlow(this)
-}

matrix-sdk-android-flow/src/test/java/org/matrix/android/sdk/flow/ExampleUnitTest.kt

@@ -1,33 +0,0 @@
-/*
- * 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 org.matrix.android.sdk.flow
-
-import org.junit.Test
-
-import org.junit.Assert.*
-
-/**
- * Example local unit test, which will execute on the development machine (host).
- *
- * See [testing documentation](http://d.android.com/tools/testing).
- */
-class ExampleUnitTest {
-    @Test
-    fun addition_isCorrect() {
-        assertEquals(4, 2 + 2)
-    }
-}

settings.gradle

@@ -5,4 +5,3 @@ include ':diff-match-patch'
 include ':attachment-viewer'
 include ':multipicker'
 include ':library:ui-styles'
-include ':matrix-sdk-android-flow'

vector/build.gradle

@@ -323,7 +323,6 @@ dependencies {
 
     implementation project(":matrix-sdk-android")
     implementation project(":matrix-sdk-android-rx")
-    implementation project(":matrix-sdk-android-flow")
     implementation project(":diff-match-patch")
     implementation project(":multipicker")
     implementation project(":attachment-viewer")

vector/src/main/java/im/vector/app/core/platform/VectorViewModel.kt

@@ -16,16 +16,10 @@
 
 package im.vector.app.core.platform
 
-import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.BaseMvRxViewModel
-import com.airbnb.mvrx.Fail
-import com.airbnb.mvrx.Loading
 import com.airbnb.mvrx.MavericksState
-import com.airbnb.mvrx.Success
 import im.vector.app.core.utils.DataSource
 import im.vector.app.core.utils.PublishDataSource
-import io.reactivex.Observable
-import io.reactivex.Single
 
 abstract class VectorViewModel<S : MavericksState, VA : VectorViewModelAction, VE : VectorViewEvents>(initialState: S)
     : BaseMvRxViewModel<S>(initialState) {
@@ -38,31 +32,5 @@ abstract class VectorViewModel<S : MavericksState, VA : VectorViewModelAction, V
     protected val _viewEvents = PublishDataSource<VE>()
     val viewEvents: DataSource<VE> = _viewEvents
 
-    /**
-     * This method does the same thing as the execute function, but it doesn't subscribe to the stream
-     * so you can use this in a switchMap or a flatMap
-     */
-    // False positive
-    @Suppress("USELESS_CAST", "NULLABLE_TYPE_PARAMETER_AGAINST_NOT_NULL_TYPE_PARAMETER")
-    fun <T> Single<T>.toAsync(stateReducer: S.(Async<T>) -> S): Single<Async<T>> {
-        setState { stateReducer(Loading()) }
-        return map { Success(it) as Async<T> }
-                .onErrorReturn { Fail(it) }
-                .doOnSuccess { setState { stateReducer(it) } }
-    }
-
-    /**
-     * This method does the same thing as the execute function, but it doesn't subscribe to the stream
-     * so you can use this in a switchMap or a flatMap
-     */
-    // False positive
-    @Suppress("USELESS_CAST", "NULLABLE_TYPE_PARAMETER_AGAINST_NOT_NULL_TYPE_PARAMETER")
-    fun <T> Observable<T>.toAsync(stateReducer: S.(Async<T>) -> S): Observable<Async<T>> {
-        setState { stateReducer(Loading()) }
-        return map { Success(it) as Async<T> }
-                .onErrorReturn { Fail(it) }
-                .doOnNext { setState { stateReducer(it) } }
-    }
-
     abstract fun handle(action: VA)
 }

vector/src/main/java/im/vector/app/core/utils/OptionalFlow.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.matrix.android.sdk.flow
+package im.vector.app.core.utils
 
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.filter

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.crypto.quads
 
+import androidx.lifecycle.asFlow
 import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
@@ -43,9 +44,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
 import java.io.ByteArrayOutputStream
 
@@ -116,8 +115,9 @@ class SharedSecureStorageViewModel @AssistedInject constructor(
             }
         }
 
-        session.flow()
+        session.cryptoService()
-                .liveUserCryptoDevices(session.myUserId)
+                .getLiveCryptoDeviceInfo(session.myUserId)
+                .asFlow()
                 .distinctUntilChanged()
                 .execute {
                     copy(

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.devtools
 
+import androidx.lifecycle.asFlow
 import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
@@ -40,9 +41,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
 
 class RoomDevToolViewModel @AssistedInject constructor(
         @Assisted val initialState: RoomDevToolViewState,
@@ -70,8 +69,8 @@ class RoomDevToolViewModel @AssistedInject constructor(
 
     init {
         session.getRoom(initialState.roomId)
-                ?.flow()
+                ?.getStateEventsLive(emptySet())
-                ?.liveStateEvents(emptySet())
+                ?.asFlow()
                 ?.execute { async ->
                     copy(stateEvents = async)
                 }

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

@@ -15,6 +15,7 @@
  */
 package im.vector.app.features.discovery
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -36,7 +37,6 @@ 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.flow.flow
 
 class DiscoverySettingsViewModel @AssistedInject constructor(
         @Assisted initialState: DiscoverySettingsState,
@@ -85,8 +85,9 @@ class DiscoverySettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeThreePids() {
-        session.flow()
+        session
-                .liveThreePIds(true)
+                .getThreePidsLive(true)
+                .asFlow()
                 .onEach {
                     retrieveBinding(it)
                 }

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

@@ -47,12 +47,9 @@ 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
-import org.matrix.android.sdk.rx.asObservable
-import org.matrix.android.sdk.rx.rx
 import timber.log.Timber
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
@@ -104,8 +101,10 @@ class HomeActivityViewModel @AssistedInject constructor(
                 .crossSigningService().allPrivateKeysKnown()
 
         safeActiveSession
-                .flow()
+                .cryptoService()
-                .liveCrossSigningInfo(safeActiveSession.myUserId)
+                .crossSigningService()
+                .getLiveCrossSigningKeys(safeActiveSession.myUserId)
+                .asFlow()
                 .onEach {
                     val isVerified = it.getOrNull()?.isTrusted() ?: false
                     if (!isVerified && onceTrusted) {

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

@@ -48,7 +48,6 @@ import org.matrix.android.sdk.api.session.room.RoomSortOrder
 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.rx.asObservable
 import timber.log.Timber
 import java.util.concurrent.TimeUnit
@@ -96,11 +95,13 @@ class HomeDetailViewModel @AssistedInject constructor(@Assisted initialState: Ho
         updateShowDialPadTab()
         observeDataStore()
         callManager.addProtocolsCheckerListener(this)
-        session.flow().liveUser(session.myUserId).execute {
+        session.getUserLive(session.myUserId)
-            copy(
+                .asFlow()
-                    myMatrixItem = it.invoke()?.getOrNull()?.toMatrixItem()
+                .execute {
-            )
+                    copy(
-        }
+                            myMatrixItem = it.invoke()?.getOrNull()?.toMatrixItem()
+                    )
+                }
     }
 
     private fun observeDataStore() {
@@ -183,8 +184,8 @@ class HomeDetailViewModel @AssistedInject constructor(@Assisted initialState: Ho
     }
 
     private fun observeSyncState() {
-        session.flow()
+        session.getSyncStateLive()
-                .liveSyncState()
+                .asFlow()
                 .setOnEach { syncState ->
                     copy(syncState = syncState)
                 }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.home
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -41,7 +42,6 @@ import org.matrix.android.sdk.api.extensions.orFalse
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.util.MatrixItem
 import org.matrix.android.sdk.api.util.toMatrixItem
-import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.model.rest.DeviceInfo
 import timber.log.Timber
 
@@ -99,9 +99,9 @@ class UnknownDeviceDetectorSharedViewModel @AssistedInject constructor(@Assisted
         )
 
         combine(
-                session.flow().liveUserCryptoDevices(session.myUserId),
+                session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow(),
-                session.flow().liveMyDevicesInfo(),
+                session.cryptoService().getLiveMyDevicesInfo().asFlow(),
-                session.flow().liveCrossSigningPrivateKeys()
+                session.cryptoService().crossSigningService().getLiveCrossSigningPrivateKeys().asFlow()
         )
         { cryptoList, infoList, pInfo ->
             //                    Timber.v("## Detector trigger ${cryptoList.map { "${it.deviceId} ${it.trustLevel}" }}")
@@ -132,7 +132,7 @@ class UnknownDeviceDetectorSharedViewModel @AssistedInject constructor(@Assisted
                     )
                 }
 
-        session.flow().liveUserCryptoDevices(session.myUserId)
+        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow()
                 .distinctUntilChanged()
                 .sample(5_000)
                 .onEach {

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.home.room.breadcrumbs
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -25,13 +26,10 @@ import dagger.assisted.AssistedFactory
 import im.vector.app.core.platform.EmptyAction
 import im.vector.app.core.platform.EmptyViewEvents
 import im.vector.app.core.platform.VectorViewModel
-import io.reactivex.schedulers.Schedulers
 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,
                                                        private val session: Session)
@@ -62,11 +60,11 @@ class BreadcrumbsViewModel @AssistedInject constructor(@Assisted initialState: B
     // PRIVATE METHODS *****************************************************************************
 
     private fun observeBreadcrumbs() {
-        session.flow()
+        session.getBreadcrumbsLive(roomSummaryQueryParams {
-                .liveBreadcrumbs(roomSummaryQueryParams {
                     displayName = QueryStringValue.NoCondition
                     memberships = listOf(Membership.JOIN)
                 })
+                .asFlow()
                 .execute { asyncBreadcrumbs ->
                     copy(asyncBreadcrumbs = asyncBreadcrumbs)
                 }

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

@@ -37,6 +37,7 @@ import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.mvrx.runCatchingToAsync
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.attachments.toContentAttachmentData
 import im.vector.app.features.call.conference.ConferenceEvent
 import im.vector.app.features.call.conference.JitsiActiveConferenceHolder
@@ -108,8 +109,6 @@ 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 timber.log.Timber
 import java.util.concurrent.TimeUnit
@@ -254,11 +253,12 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeActiveRoomWidgets() {
-        session.flow()
+        session.widgetService()
-                .liveRoomWidgets(
+                .getRoomWidgetsLive(
                         roomId = initialState.roomId,
                         widgetId = QueryStringValue.NoCondition
                 )
+                .asFlow()
                 .map { widgets ->
                     widgets.filter { it.isActive }
                 }
@@ -287,8 +287,9 @@ class RoomDetailViewModel @AssistedInject constructor(
         val queryParams = roomMemberQueryParams {
             this.userId = QueryStringValue.Equals(session.myUserId, QueryStringValue.Case.SENSITIVE)
         }
-        room.flow()
+        room
-                .liveRoomMembers(queryParams)
+                .getRoomMembersLive(queryParams)
+                .asFlow()
                 .map {
                     it.firstOrNull().toOptional()
                 }
@@ -1505,8 +1506,8 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeSyncState() {
-        session.flow()
+        session.getSyncStateLive()
-                .liveSyncState()
+                .asFlow()
                 .setOnEach { syncState ->
                     copy(syncState = syncState)
                 }
@@ -1520,7 +1521,8 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeRoomSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(
@@ -1532,7 +1534,7 @@ class RoomDetailViewModel @AssistedInject constructor(
     private fun getUnreadState() {
         combine(
                 timelineEvents,
-                room.flow().liveRoomSummary().unwrap()
+                room.getRoomSummaryLive().asFlow().unwrap()
         ) { timelineEvents, roomSummary ->
             computeUnreadState(timelineEvents, roomSummary)
         }
@@ -1579,8 +1581,8 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.flow()
+        session.getChangeMembershipsLive()
-                .liveRoomChangeMembershipState()
+                .asFlow()
                 .map {
                     it[initialState.roomId] ?: ChangeMembershipState.Unknown
                 }

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

@@ -15,6 +15,7 @@
  */
 package im.vector.app.features.home.room.detail.timeline.action
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -28,6 +29,7 @@ import im.vector.app.core.extensions.canReact
 import im.vector.app.core.platform.EmptyViewEvents
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.home.room.detail.timeline.format.NoticeEventFormatter
 import im.vector.app.features.html.EventHtmlRenderer
 import im.vector.app.features.html.PillsPostProcessor
@@ -58,8 +60,6 @@ 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
 
 /**
  * Information related to an event and used to display preview in contextual bottom sheet.
@@ -137,8 +137,8 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
 
     private fun observeEvent() {
         if (room == null) return
-        room.flow()
+        room.getTimeLineEventLive(initialState.eventId)
-                .liveTimelineEvent(initialState.eventId)
+                .asFlow()
                 .unwrap()
                 .execute {
                     copy(timelineEvent = it)
@@ -149,8 +149,8 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
         if (room == null) return
         eventIdFlow
                 .flatMapLatest { eventId ->
-                    room.flow()
+                    room.getEventAnnotationsSummaryLive(eventId)
-                            .liveAnnotationSummary(eventId)
+                            .asFlow()
                             .map { annotations ->
                                 EmojiDataSource.quickEmojis.map { emoji ->
                                     ToggleState(emoji, annotations.getOrNull()?.reactionsSummary?.firstOrNull { it.key == emoji }?.addedByMe ?: false)

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

@@ -17,6 +17,7 @@
 package im.vector.app.features.home.room.list
 
 import androidx.lifecycle.MutableLiveData
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -42,7 +43,6 @@ 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.flow.flow
 import timber.log.Timber
 import javax.inject.Inject
 
@@ -95,7 +95,8 @@ class RoomListViewModel @Inject constructor(
                     )
                 }
 
-        session.flow().liveUser(session.myUserId)
+        session.getUserLive(session.myUserId)
+                .asFlow()
                 .map { it.getOrNull()?.getBestName() }
                 .distinctUntilChanged()
                 .execute {
@@ -106,8 +107,8 @@ class RoomListViewModel @Inject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.flow()
+        session.getChangeMembershipsLive()
-                .liveRoomChangeMembershipState()
+                .asFlow()
                 .setOnEach {
                     copy(roomMembershipChanges = it)
                 }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.invite
 
+import androidx.lifecycle.asFlow
 import im.vector.app.ActiveSessionDataSource
 import im.vector.app.features.session.coroutineScope
 import io.reactivex.disposables.Disposable
@@ -35,7 +36,6 @@ 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.flow.flow
 import timber.log.Timber
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -82,10 +82,9 @@ class InvitesAcceptor @Inject constructor(
         val roomQueryParams = roomSummaryQueryParams {
             this.memberships = listOf(Membership.INVITE)
         }
-        val flowSession = session.flow()
         combine(
-                flowSession.liveRoomSummaries(roomQueryParams),
+                session.getRoomSummariesLive(roomQueryParams).asFlow(),
-                flowSession.liveRoomChangeMembershipState().debounce(1000)
+                session.getChangeMembershipsLive().asFlow().debounce(1000)
         ) { invitedRooms, _ -> invitedRooms.map { it.roomId } }
                 .filter { it.isNotEmpty() }
                 .onEach { invitedRoomIds ->

vector/src/main/java/im/vector/app/features/login2/created/AccountCreatedViewModel.kt

@@ -16,7 +16,7 @@
 
 package im.vector.app.features.login2.created
 
-import androidx.lifecycle.viewModelScope
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -24,13 +24,13 @@ import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import im.vector.app.core.platform.VectorViewModel
+import im.vector.app.core.utils.unwrap
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.MatrixPatterns
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.util.MatrixItem
 import org.matrix.android.sdk.api.util.toMatrixItem
-import org.matrix.android.sdk.rx.rx
-import org.matrix.android.sdk.rx.unwrap
 import timber.log.Timber
 
 class AccountCreatedViewModel @AssistedInject constructor(
@@ -62,8 +62,8 @@ class AccountCreatedViewModel @AssistedInject constructor(
     }
 
     private fun observeUser() {
-        session.rx()
+        session.getUserLive(session.myUserId)
-                .liveUser(session.myUserId)
+                .asFlow()
                 .unwrap()
                 .map {
                     if (MatrixPatterns.isUserId(it.userId)) {

vector/src/main/java/im/vector/app/features/powerlevel/PowerLevelsFlowFactory.kt

@@ -16,6 +16,9 @@
 
 package im.vector.app.features.powerlevel
 
+import androidx.lifecycle.asFlow
+import im.vector.app.core.utils.mapOptional
+import im.vector.app.core.utils.unwrap
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flowOn
@@ -24,15 +27,13 @@ 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.Room
 import org.matrix.android.sdk.api.session.room.model.PowerLevelsContent
-import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.flow.mapOptional
-import org.matrix.android.sdk.flow.unwrap
 
 class PowerLevelsFlowFactory(private val room: Room) {
 
     fun createFlow(): Flow<PowerLevelsContent> {
-        return room.flow()
+        return room
-                .liveStateEvent(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+                .getStateEventLive(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+                .asFlow()
                 .flowOn(Dispatchers.Default)
                 .mapOptional { it.content.toModel<PowerLevelsContent>() }
                 .unwrap()

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.room
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
@@ -27,6 +28,7 @@ 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 im.vector.app.core.utils.unwrap
 import io.reactivex.Observable
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
@@ -47,8 +49,6 @@ 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.flow.flow
-import org.matrix.android.sdk.flow.unwrap
 
 /**
  * This ViewModel observe a room summary and notify when the room is left
@@ -90,8 +90,8 @@ class RequireActiveMembershipViewModel @AssistedInject constructor(
                         val emptyResult = Optional.empty<RequireActiveMembershipViewEvents.RoomLeft>()
                         emit(emptyResult)
                     }
-                    room.flow()
+                    room.getRoomSummaryLive()
-                            .liveRoomSummary()
+                            .asFlow()
                             .unwrap()
                             .flowOn(Dispatchers.Default)
                             .map { mapToLeftViewEvent(room, it) }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.roomdirectory
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.Loading
@@ -38,7 +39,6 @@ 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.flow.flow
 import timber.log.Timber
 
 class RoomDirectoryViewModel @AssistedInject constructor(
@@ -80,8 +80,8 @@ class RoomDirectoryViewModel @AssistedInject constructor(
             memberships = listOf(Membership.JOIN)
         }
         session
-                .flow()
+                .getRoomSummariesLive(queryParams)
-                .liveRoomSummaries(queryParams)
+                .asFlow()
                 .map { roomSummaries ->
                     roomSummaries
                             .map { it.roomId }
@@ -93,8 +93,8 @@ class RoomDirectoryViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.flow()
+        session.getChangeMembershipsLive()
-                .liveRoomChangeMembershipState()
+                .asFlow()
                 .setOnEach {
                     copy(changeMembershipStates = it)
                 }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.roomdirectory.roompreview
 
+import androidx.lifecycle.asFlow
 import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
@@ -42,8 +43,6 @@ 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
 
 class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val initialState: RoomPreviewViewState,
@@ -168,8 +167,8 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
             excludeType = null
         }
         session
-                .flow()
+                .getRoomSummariesLive(queryParams)
-                .liveRoomSummaries(queryParams)
+                .asFlow()
                 .onEach { list ->
                     val isRoomJoined = list.any {
                         it.membership == Membership.JOIN
@@ -187,8 +186,8 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
     }
 
     private fun observeMembershipChanges() {
-        session.flow()
+        session.getChangeMembershipsLive()
-                .liveRoomChangeMembershipState()
+                .asFlow()
                 .onEach {
                     val changeMembership = it[initialState.roomId] ?: ChangeMembershipState.Unknown
                     val joinState = when (changeMembership) {

vector/src/main/java/im/vector/app/features/roommemberprofile/RoomMemberProfileViewModel.kt

@@ -17,6 +17,7 @@
 
 package im.vector.app.features.roommemberprofile
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
@@ -31,6 +32,7 @@ import im.vector.app.R
 import im.vector.app.core.mvrx.runCatchingToAsync
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.combine
@@ -53,8 +55,6 @@ import org.matrix.android.sdk.api.session.room.powerlevels.Role
 import org.matrix.android.sdk.api.util.MatrixItem
 import org.matrix.android.sdk.api.util.toMatrixItem
 import org.matrix.android.sdk.api.util.toOptional
-import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.flow.unwrap
 
 class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private val initialState: RoomMemberProfileViewState,
                                                              private val stringProvider: StringProvider,
@@ -107,7 +107,7 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
             }
         }
 
-        session.flow().liveUserCryptoDevices(initialState.userId)
+        session.cryptoService().getLiveCryptoDeviceInfo(initialState.userId).asFlow()
                 .map {
                     Pair(
                             it.fold(true, { prev, dev -> prev && dev.isVerified }),
@@ -121,14 +121,14 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
                     )
                 }
 
-        session.flow().liveCrossSigningInfo(initialState.userId)
+        session.cryptoService().crossSigningService().getLiveCrossSigningKeys(initialState.userId).asFlow()
                 .execute {
                     copy(userMXCrossSigningInfo = it.invoke()?.getOrNull())
                 }
     }
 
     private fun observeIgnoredState() {
-        session.flow().liveIgnoredUsers()
+        session.getIgnoredUsersLive().asFlow()
                 .map { ignored ->
                     ignored.find {
                         it.userId == initialState.userId
@@ -245,7 +245,7 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
         val queryParams = roomMemberQueryParams {
             this.userId = QueryStringValue.Equals(initialState.userId, QueryStringValue.Case.SENSITIVE)
         }
-        room.flow().liveRoomMembers(queryParams)
+        room.getRoomMembersLive(queryParams).asFlow()
                 .map { it.firstOrNull().toOptional() }
                 .unwrap()
                 .execute {
@@ -285,7 +285,7 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
     }
 
     private fun observeRoomSummaryAndPowerLevels(room: Room) {
-        val roomSummaryLive = room.flow().liveRoomSummary().unwrap()
+        val roomSummaryLive = room.getRoomSummaryLive().asFlow().unwrap()
         val powerLevelsContentLive = PowerLevelsFlowFactory(room).createFlow()
 
         powerLevelsContentLive

vector/src/main/java/im/vector/app/features/roommemberprofile/devices/DeviceListBottomSheetViewModel.kt

@@ -16,6 +16,7 @@
  */
 package im.vector.app.features.roommemberprofile.devices
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
@@ -33,9 +34,7 @@ import org.matrix.android.sdk.api.session.crypto.crosssigning.MXCrossSigningInfo
 import org.matrix.android.sdk.api.session.crypto.verification.VerificationMethod
 import org.matrix.android.sdk.api.util.MatrixItem
 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.rx.rx
 
 data class DeviceListViewState(
         val userItem: MatrixItem? = null,
@@ -56,14 +55,17 @@ class DeviceListBottomSheetViewModel @AssistedInject constructor(@Assisted priva
     }
 
     init {
-        session.flow().liveUserCryptoDevices(args.userId)
+        session.cryptoService().getLiveCryptoDeviceInfo(args.userId)
+                .asFlow()
                 .execute {
                     copy(cryptoDevices = it).also {
                         refreshSelectedId()
                     }
                 }
 
-        session.flow().liveCrossSigningInfo(args.userId)
+        session.cryptoService().crossSigningService()
+                .getLiveCrossSigningKeys(args.userId)
+                .asFlow()
                 .execute {
                     copy(memberCrossSigningKey = it.invoke()?.getOrNull())
                 }

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

@@ -17,6 +17,7 @@
 
 package im.vector.app.features.roomprofile
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -27,6 +28,8 @@ 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 im.vector.app.core.utils.mapOptional
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.home.ShortcutCreator
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import kotlinx.coroutines.Dispatchers
@@ -40,10 +43,6 @@ 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.flow.FlowRoom
-import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.flow.mapOptional
-import org.matrix.android.sdk.flow.unwrap
 
 class RoomProfileViewModel @AssistedInject constructor(
         @Assisted private val initialState: RoomProfileViewState,
@@ -69,15 +68,14 @@ class RoomProfileViewModel @AssistedInject constructor(
     private val room = session.getRoom(initialState.roomId)!!
 
     init {
-        val flowRoom = room.flow()
+        observeRoomSummary()
-        observeRoomSummary(flowRoom)
+        observeRoomCreateContent()
-        observeRoomCreateContent(flowRoom)
+        observeBannedRoomMembers()
-        observeBannedRoomMembers(flowRoom)
         observePermissions()
     }
 
-    private fun observeRoomCreateContent(flowRoom: FlowRoom) {
+    private fun observeRoomCreateContent() {
-        flowRoom.liveStateEvent(EventType.STATE_ROOM_CREATE, QueryStringValue.NoCondition)
+        room.getStateEventLive(EventType.STATE_ROOM_CREATE, QueryStringValue.NoCondition).asFlow()
                 .mapOptional { it.content.toModel<RoomCreateContent>() }
                 .unwrap()
                 .execute { async ->
@@ -92,16 +90,16 @@ class RoomProfileViewModel @AssistedInject constructor(
                 }
     }
 
-    private fun observeRoomSummary(flowRoom: FlowRoom) {
+    private fun observeRoomSummary() {
-        flowRoom.liveRoomSummary()
+        room.getRoomSummaryLive().asFlow()
                 .unwrap()
                 .execute {
                     copy(roomSummary = it)
                 }
     }
 
-    private fun observeBannedRoomMembers(flowRoom: FlowRoom) {
+    private fun observeBannedRoomMembers() {
-        flowRoom.liveRoomMembers(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
+        room.getRoomMembersLive(roomMemberQueryParams { memberships = listOf(Membership.BAN) }).asFlow()
                 .execute {
                     copy(bannedMembership = it)
                 }

vector/src/main/java/im/vector/app/features/roomprofile/alias/RoomAliasViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.roomprofile.alias
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
@@ -28,6 +29,8 @@ 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.core.utils.mapOptional
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
@@ -39,11 +42,6 @@ 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.RoomCanonicalAliasContent
 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
-import org.matrix.android.sdk.rx.unwrap
 
 class RoomAliasViewModel @AssistedInject constructor(@Assisted initialState: RoomAliasViewState,
                                                      private val session: Session)
@@ -131,7 +129,8 @@ class RoomAliasViewModel @AssistedInject constructor(@Assisted initialState: Roo
     }
 
     private fun observeRoomSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(
@@ -173,8 +172,8 @@ class RoomAliasViewModel @AssistedInject constructor(@Assisted initialState: Roo
      * We do not want to use the fallback avatar url, which can be the other user avatar, or the current user avatar.
      */
     private fun observeRoomCanonicalAlias() {
-        room.flow()
+        room.getStateEventLive(EventType.STATE_ROOM_CANONICAL_ALIAS, QueryStringValue.NoCondition)
-                .liveStateEvent(EventType.STATE_ROOM_CANONICAL_ALIAS, QueryStringValue.NoCondition)
+                .asFlow()
                 .mapOptional { it.content.toModel<RoomCanonicalAliasContent>() }
                 .unwrap()
                 .setOnEach {

vector/src/main/java/im/vector/app/features/roomprofile/banned/RoomBannedMemberListViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.roomprofile.banned
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -26,6 +27,7 @@ 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 im.vector.app.core.utils.unwrap
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
@@ -38,10 +40,6 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.RoomMemberContent
 import org.matrix.android.sdk.api.session.room.model.RoomMemberSummary
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
-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
 
 class RoomBannedMemberListViewModel @AssistedInject constructor(@Assisted initialState: RoomBannedMemberListViewState,
                                                                 private val stringProvider: StringProvider,
@@ -57,13 +55,15 @@ class RoomBannedMemberListViewModel @AssistedInject constructor(@Assisted initia
 
     init {
 
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)
                 }
 
-        room.flow().liveRoomMembers(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
+        room.getRoomMembersLive(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
+                .asFlow()
                 .execute {
                     copy(
                             bannedMemberSummaries = it

vector/src/main/java/im/vector/app/features/roomprofile/members/RoomMemberListViewModel.kt

@@ -27,6 +27,8 @@ 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 im.vector.app.core.utils.mapOptional
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import io.reactivex.android.schedulers.AndroidSchedulers
 import kotlinx.coroutines.Dispatchers
@@ -51,9 +53,6 @@ import org.matrix.android.sdk.api.session.room.model.PowerLevelsContent
 import org.matrix.android.sdk.api.session.room.model.RoomMemberSummary
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.api.session.room.powerlevels.Role
-import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.flow.mapOptional
-import org.matrix.android.sdk.flow.unwrap
 import timber.log.Timber
 
 class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState: RoomMemberListViewState,
@@ -94,9 +93,9 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
         }
 
         combine(
-                room.flow().liveRoomMembers(roomMemberQueryParams),
+                room.getRoomMembersLive(roomMemberQueryParams).asFlow(),
-                room.flow()
+                room.getStateEventLive(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
-                        .liveStateEvent(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+                        .asFlow()
                         .mapOptional { it.content.toModel<PowerLevelsContent>() }
                         .unwrap()
         )
@@ -109,7 +108,8 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
                 }
 
         if (room.isEncrypted()) {
-            room.flow().liveRoomMembers(roomMemberQueryParams)
+            room.getRoomMembersLive(roomMemberQueryParams)
+                    .asFlow()
                     .flowOn(Dispatchers.Main)
                     .flatMapLatest { membersSummary ->
                         session.cryptoService().getLiveCryptoDeviceInfo(membersSummary.map { it.userId })
@@ -153,7 +153,8 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
     }
 
     private fun observeRoomSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)
@@ -161,7 +162,8 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
     }
 
     private fun observeThirdPartyInvites() {
-        room.flow().liveStateEvents(setOf(EventType.STATE_ROOM_THIRD_PARTY_INVITE))
+        room.getStateEventsLive(setOf(EventType.STATE_ROOM_THIRD_PARTY_INVITE))
+                .asFlow()
                 .execute { async ->
                     copy(threePidInvites = async)
                 }

vector/src/main/java/im/vector/app/features/roomprofile/notifications/RoomNotificationSettingsViewModel.kt

@@ -16,7 +16,7 @@
 
 package im.vector.app.features.roomprofile.notifications
 
-import androidx.lifecycle.viewModelScope
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.Success
@@ -25,13 +25,10 @@ import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import im.vector.app.core.platform.VectorViewModel
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.home.room.list.actions.RoomListQuickActionsBottomSheet
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
-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
 
 class RoomNotificationSettingsViewModel @AssistedInject constructor(
         @Assisted initialState: RoomNotificationSettingsViewState,
@@ -66,7 +63,8 @@ class RoomNotificationSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)
@@ -74,8 +72,8 @@ class RoomNotificationSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeNotificationState() {
-        room.rx()
+        room.getLiveRoomNotificationState()
-                .liveNotificationState()
+                .asFlow()
                 .execute {
                     copy(notificationState = it)
                 }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.roomprofile.permissions
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.Success
@@ -25,6 +26,7 @@ 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.core.utils.unwrap
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
@@ -34,8 +36,6 @@ 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.flow.flow
-import org.matrix.android.sdk.flow.unwrap
 
 class RoomPermissionsViewModel @AssistedInject constructor(@Assisted initialState: RoomPermissionsViewState,
                                                            private val session: Session)
@@ -63,7 +63,8 @@ class RoomPermissionsViewModel @AssistedInject constructor(@Assisted initialStat
     }
 
     private fun observeRoomSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(

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

@@ -17,6 +17,7 @@
 package im.vector.app.features.roomprofile.settings
 
 import androidx.core.net.toFile
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
@@ -26,6 +27,8 @@ 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.core.utils.mapOptional
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.powerlevel.PowerLevelsFlowFactory
 import im.vector.app.features.settings.VectorPreferences
 import kotlinx.coroutines.flow.launchIn
@@ -43,9 +46,6 @@ import org.matrix.android.sdk.api.session.room.model.RoomGuestAccessContent
 import org.matrix.android.sdk.api.session.room.model.RoomHistoryVisibilityContent
 import org.matrix.android.sdk.api.session.room.model.RoomJoinRulesContent
 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
 
 class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState: RoomSettingsViewState,
                                                         private val vectorPreferences: VectorPreferences,
@@ -125,7 +125,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeRoomSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     val roomSummary = async.invoke()
@@ -161,8 +162,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeRoomHistoryVisibility() {
-        room.flow()
+        room.getStateEventLive(EventType.STATE_ROOM_HISTORY_VISIBILITY, QueryStringValue.NoCondition)
-                .liveStateEvent(EventType.STATE_ROOM_HISTORY_VISIBILITY, QueryStringValue.NoCondition)
+                .asFlow()
                 .mapOptional { it.content.toModel<RoomHistoryVisibilityContent>() }
                 .unwrap()
                 .mapNotNull { it.historyVisibility }
@@ -172,8 +173,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeJoinRule() {
-        room.flow()
+        room.getStateEventLive(EventType.STATE_ROOM_JOIN_RULES, QueryStringValue.NoCondition)
-                .liveStateEvent(EventType.STATE_ROOM_JOIN_RULES, QueryStringValue.NoCondition)
+                .asFlow()
                 .mapOptional { it.content.toModel<RoomJoinRulesContent>() }
                 .unwrap()
                 .mapNotNull { it.joinRules }
@@ -183,8 +184,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeGuestAccess() {
-        room.flow()
+        room.getStateEventLive(EventType.STATE_ROOM_GUEST_ACCESS, QueryStringValue.NoCondition)
-                .liveStateEvent(EventType.STATE_ROOM_GUEST_ACCESS, QueryStringValue.NoCondition)
+                .asFlow()
                 .mapOptional { it.content.toModel<RoomGuestAccessContent>() }
                 .unwrap()
                 .mapNotNull { it.guestAccess }
@@ -197,8 +198,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
      * We do not want to use the fallback avatar url, which can be the other user avatar, or the current user avatar.
      */
     private fun observeRoomAvatar() {
-        room.flow()
+        room.getStateEventLive(EventType.STATE_ROOM_AVATAR, QueryStringValue.NoCondition)
-                .liveStateEvent(EventType.STATE_ROOM_AVATAR, QueryStringValue.NoCondition)
+                .asFlow()
                 .mapOptional { it.content.toModel<RoomAvatarContent>() }
                 .unwrap()
                 .setOnEach {

vector/src/main/java/im/vector/app/features/roomprofile/uploads/RoomUploadsViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.roomprofile.uploads
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -28,11 +29,10 @@ 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.core.utils.unwrap
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.room.model.message.MessageType
-import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.flow.unwrap
 
 class RoomUploadsViewModel @AssistedInject constructor(
         @Assisted initialState: RoomUploadsViewState,
@@ -65,7 +65,8 @@ class RoomUploadsViewModel @AssistedInject constructor(
     }
 
     private fun observeRoomSummary() {
-        room.flow().liveRoomSummary()
+        room.getRoomSummaryLive()
+                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.settings
 
+import androidx.lifecycle.asFlow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
@@ -25,7 +26,6 @@ import org.matrix.android.sdk.api.session.crypto.crosssigning.KEYBACKUP_SECRET_S
 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(
@@ -39,11 +39,12 @@ data class SecretsSynchronisationInfo(
 )
 
 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)),
+            accountDataService()
-            sessionFlow.liveCrossSigningInfo(myUserId),
+                    .getLiveUserAccountDataEvents(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME, KEYBACKUP_SECRET_SSSS_NAME))
-            sessionFlow.liveCrossSigningPrivateKeys()
+                    .asFlow(),
+            cryptoService().crossSigningService().getLiveCrossSigningKeys(myUserId).asFlow(),
+            cryptoService().crossSigningService().getLiveCrossSigningPrivateKeys().asFlow()
     ) { _, crossSigningInfo, pInfo ->
         // first check if 4S is already setup
         val is4SSetup = sharedSecretStorageService.isRecoverySetup()

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

@@ -26,6 +26,7 @@ import android.view.ViewGroup
 import android.widget.Toast
 import androidx.appcompat.app.AlertDialog
 import androidx.core.view.isVisible
+import androidx.lifecycle.asFlow
 import androidx.lifecycle.lifecycleScope
 import androidx.preference.EditTextPreference
 import androidx.preference.Preference
@@ -47,6 +48,7 @@ import im.vector.app.core.resources.ColorProvider
 import im.vector.app.core.utils.TextUtils
 import im.vector.app.core.utils.getSizeOfFiles
 import im.vector.app.core.utils.toast
+import im.vector.app.core.utils.unwrap
 import im.vector.app.databinding.DialogChangePasswordBinding
 import im.vector.app.features.MainActivity
 import im.vector.app.features.MainActivityArgs
@@ -62,8 +64,6 @@ 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.flow.flow
-import org.matrix.android.sdk.flow.unwrap
 import java.io.File
 import java.util.UUID
 import javax.inject.Inject
@@ -122,8 +122,8 @@ class VectorSettingsGeneralFragment @Inject constructor(
     }
 
     private fun observeUserAvatar() {
-        session.flow()
+        session.getUserLive(session.myUserId)
-                .liveUser(session.myUserId)
+                .asFlow()
                 .unwrap()
                 .distinctUntilChangedBy { user -> user.avatarUrl }
                 .onEach {
@@ -133,8 +133,8 @@ class VectorSettingsGeneralFragment @Inject constructor(
     }
 
     private fun observeUserDisplayName() {
-        session.flow()
+        session.getUserLive(session.myUserId)
-                .liveUser(session.myUserId)
+                .asFlow()
                 .unwrap()
                 .map { it.displayName ?: "" }
                 .distinctUntilChanged()

vector/src/main/java/im/vector/app/features/settings/crosssigning/CrossSigningSettingsViewModel.kt

@@ -15,6 +15,7 @@
  */
 package im.vector.app.features.settings.crosssigning
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -37,7 +38,6 @@ import org.matrix.android.sdk.api.auth.data.LoginFlowTypes
 import org.matrix.android.sdk.api.auth.registration.RegistrationFlowResponse
 import org.matrix.android.sdk.api.auth.registration.nextUncompletedStage
 import org.matrix.android.sdk.api.session.Session
-import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.crosssigning.fromBase64
 import org.matrix.android.sdk.internal.crypto.crosssigning.isVerified
 import org.matrix.android.sdk.internal.crypto.model.rest.DefaultBaseAuth
@@ -56,8 +56,8 @@ class CrossSigningSettingsViewModel @AssistedInject constructor(
 
     init {
         combine(
-                session.flow().liveMyDevicesInfo(),
+                session.cryptoService().getLiveMyDevicesInfo().asFlow(),
-                session.flow().liveCrossSigningInfo(session.myUserId)
+                session.cryptoService().crossSigningService().getLiveCrossSigningKeys(session.myUserId).asFlow()
         )
         { myDevicesInfo, mxCrossSigningInfo ->
             myDevicesInfo to mxCrossSigningInfo

vector/src/main/java/im/vector/app/features/settings/devices/DeviceVerificationInfoBottomSheetViewModel.kt

@@ -15,6 +15,7 @@
  */
 package im.vector.app.features.settings.devices
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.Loading
 import com.airbnb.mvrx.MavericksViewModelFactory
@@ -27,9 +28,7 @@ import im.vector.app.core.platform.EmptyViewEvents
 import im.vector.app.core.platform.VectorViewModel
 import kotlinx.coroutines.flow.map
 import org.matrix.android.sdk.api.session.Session
-import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.model.rest.DeviceInfo
-import org.matrix.android.sdk.rx.rx
 
 class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@Assisted initialState: DeviceVerificationInfoBottomSheetViewState,
                                                                              @Assisted val deviceId: String,
@@ -50,7 +49,8 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
                     isRecoverySetup = session.sharedSecretStorageService.isRecoverySetup()
             )
         }
-        session.flow().liveCrossSigningInfo(session.myUserId)
+        session.cryptoService().crossSigningService().getLiveCrossSigningKeys(session.myUserId)
+                .asFlow()
                 .execute {
                     copy(
                             hasAccountCrossSigning = it.invoke()?.getOrNull() != null,
@@ -58,7 +58,8 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
                     )
                 }
 
-        session.flow().liveUserCryptoDevices(session.myUserId)
+        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId)
+                .asFlow()
                 .map { list ->
                     list.firstOrNull { it.deviceId == deviceId }
                 }
@@ -69,7 +70,7 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
                     )
                 }
 
-        session.flow().liveUserCryptoDevices(session.myUserId)
+        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow()
                 .map { it.size }
                 .execute {
                     copy(
@@ -81,7 +82,8 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
             copy(deviceInfo = Loading())
         }
 
-        session.flow().liveMyDevicesInfo()
+        session.cryptoService().getLiveMyDevicesInfo()
+                .asFlow()
                 .map { devices ->
                     devices.firstOrNull { it.deviceId == deviceId } ?: DeviceInfo(deviceId = deviceId)
                 }

vector/src/main/java/im/vector/app/features/settings/devices/DevicesViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.settings.devices
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -56,7 +57,6 @@ import org.matrix.android.sdk.api.session.crypto.verification.VerificationMethod
 import org.matrix.android.sdk.api.session.crypto.verification.VerificationService
 import org.matrix.android.sdk.api.session.crypto.verification.VerificationTransaction
 import org.matrix.android.sdk.api.session.crypto.verification.VerificationTxState
-import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.crosssigning.DeviceTrustLevel
 import org.matrix.android.sdk.internal.crypto.crosssigning.fromBase64
 import org.matrix.android.sdk.internal.crypto.model.CryptoDeviceInfo
@@ -123,8 +123,14 @@ class DevicesViewModel @AssistedInject constructor(
         }
 
         combine(
-                session.flow().liveUserCryptoDevices(session.myUserId),
+                session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow()
-                session.flow().liveMyDevicesInfo()
+                        .onEach {
+                            Timber.v("getLiveCryptoDeviceInfo")
+                        },
+                session.cryptoService().getLiveMyDevicesInfo().asFlow()
+                        .onEach {
+                            Timber.v("getLiveMyDevicesInfo")
+                        }
         )
         { cryptoList, infoList ->
             infoList
@@ -141,7 +147,8 @@ class DevicesViewModel @AssistedInject constructor(
                     )
                 }
 
-        session.flow().liveCrossSigningInfo(session.myUserId)
+        session.cryptoService().crossSigningService().getLiveCrossSigningKeys(session.myUserId)
+                .asFlow()
                 .execute {
                     copy(
                             hasAccountCrossSigning = it.invoke()?.getOrNull() != null,
@@ -157,7 +164,8 @@ class DevicesViewModel @AssistedInject constructor(
 //                    )
 //                }
 
-        session.flow().liveUserCryptoDevices(session.myUserId)
+        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId)
+                .asFlow()
                 .map { it.size }
                 .distinctUntilChanged()
                 .sample(5_000)

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.settings.devtools
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksState
@@ -31,7 +32,6 @@ 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.flow.flow
 
 data class AccountDataViewState(
         val accountData: Async<List<UserAccountDataEvent>> = Uninitialized
@@ -42,7 +42,9 @@ class AccountDataViewModel @AssistedInject constructor(@Assisted initialState: A
     : VectorViewModel<AccountDataViewState, AccountDataAction, EmptyViewEvents>(initialState) {
 
     init {
-        session.flow().liveUserAccountData(emptySet())
+        session.accountDataService()
+                .getLiveUserAccountDataEvents(emptySet())
+                .asFlow()
                 .execute {
                     copy(accountData = it)
                 }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.settings.ignored
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -33,7 +34,6 @@ 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.flow.flow
 
 data class IgnoredUsersViewState(
         val ignoredUsers: List<User> = emptyList(),
@@ -67,8 +67,8 @@ class IgnoredUsersViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeIgnoredUsers() {
-        session.flow()
+        session.getIgnoredUsersLive()
-                .liveIgnoredUsers()
+                .asFlow()
                 .execute { async ->
                     copy(
                             ignoredUsers = async.invoke().orEmpty()

vector/src/main/java/im/vector/app/features/settings/threepids/ThreePidsSettingsViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.settings.threepids
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -38,7 +39,6 @@ import org.matrix.android.sdk.api.auth.UserPasswordAuth
 import org.matrix.android.sdk.api.auth.registration.RegistrationFlowResponse
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.identity.ThreePid
-import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.internal.crypto.crosssigning.fromBase64
 import org.matrix.android.sdk.internal.crypto.model.rest.DefaultBaseAuth
 import timber.log.Timber
@@ -101,8 +101,8 @@ class ThreePidsSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeThreePids() {
-        session.flow()
+        session.getThreePidsLive(true)
-                .liveThreePIds(true)
+                .asFlow()
                 .execute {
                     copy(
                             threePids = it
@@ -111,8 +111,8 @@ class ThreePidsSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observePendingThreePids() {
-        session.flow()
+        session.getPendingThreePidsLive()
-                .livePendingThreePIds()
+                .asFlow()
                 .execute {
                     copy(
                             pendingThreePids = it,

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.share
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.ViewModelContext
@@ -37,7 +38,6 @@ 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
 
 class IncomingShareViewModel @AssistedInject constructor(
         @Assisted initialState: IncomingShareViewState,
@@ -69,8 +69,8 @@ class IncomingShareViewModel @AssistedInject constructor(
         val queryParams = roomSummaryQueryParams {
             memberships = listOf(Membership.JOIN)
         }
-        session
+        session.getRoomSummariesLive(queryParams)
-                .flow().liveRoomSummaries(queryParams)
+                .asFlow()
                 .execute {
                     copy(roomSummaries = it)
                 }
@@ -86,7 +86,7 @@ class IncomingShareViewModel @AssistedInject constructor(
                         displayName = displayNameQuery
                         memberships = listOf(Membership.JOIN)
                     }
-                    session.flow().liveRoomSummaries(filterQueryParams)
+                    session.getRoomSummariesLive(filterQueryParams).asFlow()
                 }
                 .sample(300)
                 .map { it.sortedWith(breadcrumbsRoomComparator) }

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.spaces
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -42,8 +43,6 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.api.session.room.powerlevels.Role
 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
 
 class SpaceMenuViewModel @AssistedInject constructor(
@@ -78,17 +77,19 @@ class SpaceMenuViewModel @AssistedInject constructor(
 
         session.getRoom(initialState.spaceId)?.let { room ->
 
-            room.flow().liveRoomSummary().onEach {
+            room.getRoomSummaryLive()
-                it.getOrNull()?.let {
+                    .asFlow()
-                    if (it.membership == Membership.LEAVE) {
+                    .onEach {
-                        setState { copy(leavingState = Success(Unit)) }
+                        it.getOrNull()?.let {
-                        if (appStateHandler.safeActiveSpaceId() == initialState.spaceId) {
+                            if (it.membership == Membership.LEAVE) {
-                            // switch to home?
+                                setState { copy(leavingState = Success(Unit)) }
-                            appStateHandler.setCurrentSpace(null, session)
+                                if (appStateHandler.safeActiveSpaceId() == initialState.spaceId) {
+                                    // switch to home?
+                                    appStateHandler.setCurrentSpace(null, session)
+                                }
+                            }
                         }
-                    }
+                    }.launchIn(viewModelScope)
-                }
-            }.launchIn(viewModelScope)
 
             PowerLevelsFlowFactory(room)
                     .createFlow()

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

@@ -53,7 +53,6 @@ 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.util.toMatrixItem
-import org.matrix.android.sdk.flow.flow
 import org.matrix.android.sdk.rx.asObservable
 import java.util.concurrent.TimeUnit
 
@@ -82,14 +81,13 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
 
     init {
 
-        session.getUserLive(session.myUserId).asObservable()
+        session.getUserLive(session.myUserId)
-                .subscribe {
+                .asFlow()
-                    setState {
+                .setOnEach {
-                        copy(
+                    copy(
-                                myMxItem = it?.getOrNull()?.toMatrixItem()?.let { Success(it) } ?: Loading()
+                            myMxItem = it.getOrNull()?.toMatrixItem()?.let { Success(it) } ?: Loading()
-                        )
+                    )
-                    }
+                }
-                }.disposeOnClear()
 
         observeSpaceSummaries()
 //        observeSelectionState()
@@ -284,16 +282,13 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
                     null)
         }
 
-        val flowSession = session.flow()
-
         combine(
-                flowSession
+                session.getUserLive(session.myUserId)
-                        .liveUser(session.myUserId)
+                        .asFlow()
                         .map {
                             it.getOrNull()
                         },
-                flowSession
+                session.spaceService().getSpaceSummariesLive(spaceSummaryQueryParams).asFlow(),
-                        .liveSpaceSummaries(spaceSummaryQueryParams),
                 session
                         .accountDataService()
                         .getLiveRoomAccountDataEvents(setOf(RoomAccountDataTypes.EVENT_TYPE_SPACE_ORDER))
@@ -319,7 +314,8 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
         // clear local echos on update
         session.accountDataService()
                 .getLiveRoomAccountDataEvents(setOf(RoomAccountDataTypes.EVENT_TYPE_SPACE_ORDER))
-                .asObservable().execute {
+                .asFlow()
+                .execute {
                     copy(
                             spaceOrderLocalEchos = emptyMap()
                     )

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

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.spaces.explore
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -43,7 +44,6 @@ 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.flow.flow
 import timber.log.Timber
 
 class SpaceDirectoryViewModel @AssistedInject constructor(
@@ -147,9 +147,8 @@ class SpaceDirectoryViewModel @AssistedInject constructor(
             memberships = listOf(Membership.JOIN)
             excludeType = null
         }
-        session
+        session.getRoomSummariesLive(queryParams)
-                .flow()
+                .asFlow()
-                .liveRoomSummaries(queryParams)
                 .map {
                     it.map { it.roomId }.toSet()
                 }
@@ -159,8 +158,8 @@ class SpaceDirectoryViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.flow()
+        session.getChangeMembershipsLive()
-                .liveRoomChangeMembershipState()
+                .asFlow()
                 .setOnEach {
                     copy(changeMembershipStates = it)
                 }

vector/src/main/java/im/vector/app/features/spaces/leave/SpaceLeaveAdvancedViewModel.kt

@@ -16,6 +16,7 @@
 
 package im.vector.app.features.spaces.leave
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -30,6 +31,7 @@ import dagger.assisted.AssistedInject
 import im.vector.app.AppStateHandler
 import im.vector.app.core.platform.EmptyViewEvents
 import im.vector.app.core.platform.VectorViewModel
+import im.vector.app.core.utils.unwrap
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
@@ -39,8 +41,6 @@ import org.matrix.android.sdk.api.query.RoomCategoryFilter
 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.flow.unwrap
 import timber.log.Timber
 
 class SpaceLeaveAdvancedViewModel @AssistedInject constructor(
@@ -97,7 +97,8 @@ class SpaceLeaveAdvancedViewModel @AssistedInject constructor(
         val spaceSummary = session.getRoomSummary(initialState.spaceId)
         setState { copy(spaceSummary = spaceSummary) }
         session.getRoom(initialState.spaceId)?.let { room ->
-            room.flow().liveRoomSummary()
+            room.getRoomSummaryLive()
+                    .asFlow()
                     .unwrap()
                     .onEach {
                         if (it.membership == Membership.LEAVE) {

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

@@ -16,12 +16,12 @@
 
 package im.vector.app.features.userdirectory
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.Uninitialized
 import com.airbnb.mvrx.ViewModelContext
-import com.jakewharton.rxrelay2.BehaviorRelay
 import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
@@ -29,21 +29,23 @@ import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.extensions.isEmail
 import im.vector.app.core.extensions.toggle
 import im.vector.app.core.platform.VectorViewModel
-import io.reactivex.Single
+import kotlinx.coroutines.Dispatchers
-import io.reactivex.android.schedulers.AndroidSchedulers
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.debounce
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.sample
 import org.matrix.android.sdk.api.MatrixPatterns
+import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.identity.IdentityServiceListener
 import org.matrix.android.sdk.api.session.identity.ThreePid
 import org.matrix.android.sdk.api.session.profile.ProfileService
 import org.matrix.android.sdk.api.session.user.model.User
 import org.matrix.android.sdk.api.util.toMatrixItem
-import org.matrix.android.sdk.api.util.toOptional
-import org.matrix.android.sdk.rx.rx
-import java.util.concurrent.TimeUnit
-
-private typealias KnownUsersSearch = String
-private typealias DirectoryUsersSearch = String
 
 data class ThreePidUser(
         val email: String,
@@ -54,9 +56,9 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                                                     private val session: Session)
     : VectorViewModel<UserListViewState, UserListAction, UserListViewEvents>(initialState) {
 
-    private val knownUsersSearch = BehaviorRelay.create<KnownUsersSearch>()
+    private val knownUsersSearch = MutableStateFlow("")
-    private val directoryUsersSearch = BehaviorRelay.create<DirectoryUsersSearch>()
+    private val directoryUsersSearch = MutableStateFlow("")
-    private val identityServerUsersSearch = BehaviorRelay.create<String>()
+    private val identityServerUsersSearch = MutableStateFlow("")
 
     @AssistedFactory
     interface Factory {
@@ -77,11 +79,10 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private val identityServerListener = object : IdentityServiceListener {
         override fun onIdentityServerChange() {
             withState {
-                identityServerUsersSearch.accept(it.searchTerm)
+                identityServerUsersSearch.tryEmit(it.searchTerm)
+                val identityServerURL = cleanISURL(session.identityService().getCurrentIdentityServerUrl())
                 setState {
-                    copy(
+                    copy(configuredIdentityServer = identityServerURL)
-                            configuredIdentityServer = cleanISURL(session.identityService().getCurrentIdentityServerUrl())
-                    )
                 }
             }
         }
@@ -120,7 +121,7 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private fun handleISUpdateConsent(action: UserListAction.UpdateUserConsent) {
         session.identityService().setUserConsent(action.consent)
         withState {
-            identityServerUsersSearch.accept(it.searchTerm)
+            identityServerUsersSearch.tryEmit(it.searchTerm)
         }
     }
 
@@ -139,9 +140,9 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                 )
             }
         }
-        identityServerUsersSearch.accept(searchTerm)
+        identityServerUsersSearch.tryEmit(searchTerm)
-        knownUsersSearch.accept(searchTerm)
+        knownUsersSearch.tryEmit(searchTerm)
-        directoryUsersSearch.accept(searchTerm)
+        directoryUsersSearch.tryEmit(searchTerm)
     }
 
     private fun handleShareMyMatrixToLink() {
@@ -151,9 +152,9 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     }
 
     private fun handleClearSearchUsers() {
-        knownUsersSearch.accept("")
+        knownUsersSearch.tryEmit("")
-        directoryUsersSearch.accept("")
+        directoryUsersSearch.tryEmit("")
-        identityServerUsersSearch.accept("")
+        identityServerUsersSearch.tryEmit("")
         setState {
             copy(searchTerm = "")
         }
@@ -162,103 +163,82 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private fun observeUsers() = withState { state ->
         identityServerUsersSearch
                 .filter { it.isEmail() }
-                .throttleLast(300, TimeUnit.MILLISECONDS)
+                .sample(300)
-                .switchMapSingle { search ->
+                .onEach { search ->
-                    val flowSession = session.rx()
+                    executeSearchEmail(search)
-                    val stream =
+                }.launchIn(viewModelScope)
-                            flowSession.lookupThreePid(ThreePid.Email(search)).flatMap {
-                                it.getOrNull()?.let { foundThreePid ->
-                                    flowSession.getProfileInfo(foundThreePid.matrixId)
-                                            .map { json ->
-                                                ThreePidUser(
-                                                        email = search,
-                                                        user = User(
-                                                                userId = foundThreePid.matrixId,
-                                                                displayName = json[ProfileService.DISPLAY_NAME_KEY] as? String,
-                                                                avatarUrl = json[ProfileService.AVATAR_URL_KEY] as? String
-                                                        )
-                                                )
-                                            }
-                                            .onErrorResumeNext {
-                                                Single.just(ThreePidUser(email = search, user = User(foundThreePid.matrixId)))
-                                            }
-                                } ?: Single.just(ThreePidUser(email = search, user = null))
-                            }
-                    stream.toAsync {
-                        copy(matchingEmail = it)
-                    }
-                }
-                .subscribe()
-                .disposeOnClear()
 
         knownUsersSearch
-                .throttleLast(300, TimeUnit.MILLISECONDS)
+                .sample(300)
-                .observeOn(AndroidSchedulers.mainThread())
+                .flowOn(Dispatchers.Main)
-                .switchMap {
+                .flatMapLatest { search ->
-                    session.rx().livePagedUsers(it, state.excludedUserIds)
+                    session.getPagedUsersLive(search, state.excludedUserIds).asFlow()
-                }
+                }.execute {
-                .execute { async ->
+                    copy(knownUsers = it)
-                    copy(knownUsers = async)
                 }
 
         directoryUsersSearch
-                .debounce(300, TimeUnit.MILLISECONDS)
+                .debounce(300)
-                .switchMapSingle { search ->
+                .onEach { search ->
-                    val stream = if (search.isBlank()) {
+                    executeSearchDirectory(state, search)
-                        Single.just(emptyList<User>())
+                }.launchIn(viewModelScope)
-                    } else {
+    }
-                        val searchObservable = session.rx()
-                                .searchUsersDirectory(search, 50, state.excludedUserIds.orEmpty())
-                                .map { users ->
-                                    users.sortedBy { it.toMatrixItem().firstLetterOfDisplayName() }
-                                }
-                        // If it's a valid user id try to use Profile API
-                        // because directory only returns users that are in public rooms or share a room with you, where as
-                        // profile will work other federations
-                        if (!MatrixPatterns.isUserId(search)) {
-                            searchObservable
-                        } else {
-                            val profileObservable = session.rx().getProfileInfo(search)
-                                    .map { json ->
-                                        User(
-                                                userId = search,
-                                                displayName = json[ProfileService.DISPLAY_NAME_KEY] as? String,
-                                                avatarUrl = json[ProfileService.AVATAR_URL_KEY] as? String
-                                        ).toOptional()
-                                    }
-                                    .onErrorResumeNext {
-                                        // Profile API can be restricted and doesn't have to return result.
-                                        // In this case allow inviting valid user ids.
-                                        Single.just(
-                                                User(
-                                                        userId = search,
-                                                        displayName = null,
-                                                        avatarUrl = null
-                                                ).toOptional()
-                                        )
-                                    }
 
-                            Single.zip(
+    private suspend fun executeSearchEmail(search: String) {
-                                    searchObservable,
+        suspend {
-                                    profileObservable,
+            val params = listOf(ThreePid.Email(search))
-                                    { searchResults, optionalProfile ->
+            val foundThreePid = tryOrNull {
-                                        val profile = optionalProfile.getOrNull() ?: return@zip searchResults
+                session.identityService().lookUp(params).firstOrNull()
-                                        val searchContainsProfile = searchResults.any { it.userId == profile.userId }
+            }
-                                        if (searchContainsProfile) {
+            if (foundThreePid == null) {
-                                            searchResults
+                null
-                                        } else {
+            } else {
-                                            listOf(profile) + searchResults
+                try {
-                                        }
+                    val json = session.getProfile(foundThreePid.matrixId)
-                                    }
+                    ThreePidUser(
+                            email = search,
+                            user = User(
+                                    userId = foundThreePid.matrixId,
+                                    displayName = json[ProfileService.DISPLAY_NAME_KEY] as? String,
+                                    avatarUrl = json[ProfileService.AVATAR_URL_KEY] as? String
                             )
-                        }
+                    )
-                    }
+                } catch (failure: Throwable) {
-                    stream.toAsync {
+                    ThreePidUser(email = search, user = User(foundThreePid.matrixId))
-                        copy(directoryUsers = it)
-                    }
                 }
-                .subscribe()
+            }
-                .disposeOnClear()
+        }.execute {
+            copy(matchingEmail = it)
+        }
+    }
+
+    private suspend fun executeSearchDirectory(state: UserListViewState, search: String) {
+        suspend {
+            if (search.isBlank()) {
+                emptyList()
+            } else {
+                val searchResult = session
+                        .searchUsersDirectory(search, 50, state.excludedUserIds.orEmpty())
+                        .sortedBy { it.toMatrixItem().firstLetterOfDisplayName() }
+                val userProfile = if (MatrixPatterns.isUserId(search)) {
+                    val json = tryOrNull { session.getProfile(search) }
+                    User(
+                            userId = search,
+                            displayName = json?.get(ProfileService.DISPLAY_NAME_KEY) as? String,
+                            avatarUrl = json?.get(ProfileService.AVATAR_URL_KEY) as? String
+                    )
+                } else {
+                    null
+                }
+                if (userProfile == null || searchResult.any { it.userId == userProfile.userId }) {
+                    searchResult
+                } else {
+                    listOf(userProfile) + searchResult
+                }
+            }
+        }.execute {
+            copy(directoryUsers = it)
+        }
     }
 
     private fun handleSelectUser(action: UserListAction.AddPendingSelection) = withState { state ->

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

@@ -17,6 +17,7 @@
 package im.vector.app.features.widgets
 
 import android.net.Uri
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -29,6 +30,8 @@ import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
+import im.vector.app.core.utils.mapOptional
+import im.vector.app.core.utils.unwrap
 import im.vector.app.features.widgets.permissions.WidgetPermissionsHelper
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.map
@@ -42,9 +45,6 @@ import org.matrix.android.sdk.api.session.integrationmanager.IntegrationManagerS
 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.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 timber.log.Timber
 import javax.net.ssl.HttpsURLConnection
 
@@ -119,7 +119,8 @@ class WidgetViewModel @AssistedInject constructor(@Assisted val initialState: Wi
         if (room == null) {
             return
         }
-        room.flow().liveStateEvent(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+        room.getStateEventLive(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+                .asFlow()
                 .mapOptional { it.content.toModel<PowerLevelsContent>() }
                 .unwrap()
                 .map {
@@ -135,8 +136,9 @@ class WidgetViewModel @AssistedInject constructor(@Assisted val initialState: Wi
             return
         }
         val widgetId = initialState.widgetId ?: return
-        session.flow()
+        session.widgetService()
-                .liveRoomWidgets(initialState.roomId, QueryStringValue.Equals(widgetId))
+                .getRoomWidgetsLive(initialState.roomId, QueryStringValue.Equals(widgetId))
+                .asFlow()
                 .filter { it.isNotEmpty() }
                 .map { it.first() }
                 .execute {

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

@@ -15,6 +15,7 @@
  */
 package im.vector.app.features.widgets.permissions
 
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
@@ -32,7 +33,6 @@ 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.flow.flow
 import timber.log.Timber
 import java.net.URL
 
@@ -49,8 +49,9 @@ class RoomWidgetPermissionViewModel @AssistedInject constructor(@Assisted val in
 
     private fun observeWidget() {
         val widgetId = initialState.widgetId ?: return
-        session.flow()
+        session.widgetService()
-                .liveRoomWidgets(initialState.roomId, QueryStringValue.Equals(widgetId))
+                .getRoomWidgetsLive(initialState.roomId, QueryStringValue.Equals(widgetId))
+                .asFlow()
                 .filter { it.isNotEmpty() }
                 .map {
                     val widget = it.first()

vector/src/main/java/im/vector/app/features/workers/signout/ServerBackupStatusViewModel.kt

@@ -17,6 +17,7 @@
 package im.vector.app.features.workers.signout
 
 import androidx.lifecycle.MutableLiveData
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -41,7 +42,6 @@ import org.matrix.android.sdk.api.session.crypto.crosssigning.SELF_SIGNING_KEY_S
 import org.matrix.android.sdk.api.session.crypto.crosssigning.USER_SIGNING_KEY_SSSS_NAME
 import org.matrix.android.sdk.api.session.crypto.keysbackup.KeysBackupState
 import org.matrix.android.sdk.api.session.crypto.keysbackup.KeysBackupStateListener
-import org.matrix.android.sdk.flow.flow
 
 data class ServerBackupStatusViewState(
         val bannerState: Async<BannerState> = Uninitialized
@@ -92,9 +92,19 @@ class ServerBackupStatusViewModel @AssistedInject constructor(@Assisted initialS
     init {
         session.cryptoService().keysBackupService().addListener(this)
         keysBackupState.value = session.cryptoService().keysBackupService().state
-        val liveUserAccountData = session.flow().liveUserAccountData(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
+        val liveUserAccountData = session.accountDataService()
-        val liveCrossSigningInfo = session.flow().liveCrossSigningInfo(session.myUserId)
+                .getLiveUserAccountDataEvents(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
-        val liveCrossSigningPrivateKeys = session.flow().liveCrossSigningPrivateKeys()
+                .asFlow()
+        val liveCrossSigningInfo = session.cryptoService()
+                .crossSigningService()
+                .getLiveCrossSigningKeys(session.myUserId)
+                .asFlow()
+        val liveCrossSigningPrivateKeys = session.cryptoService()
+                .crossSigningService()
+                .getLiveCrossSigningPrivateKeys()
+                .asFlow()
+
+
         combine(liveUserAccountData, liveCrossSigningInfo, keyBackupFlow, liveCrossSigningPrivateKeys) { _, crossSigningInfo, keyBackupState, pInfo ->
             // first check if 4S is already setup
             if (session.sharedSecretStorageService.isRecoverySetup()) {

vector/src/main/java/im/vector/app/features/workers/signout/SignoutCheckViewModel.kt

@@ -17,6 +17,7 @@
 package im.vector.app.features.workers.signout
 
 import android.net.Uri
+import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -43,7 +44,6 @@ import org.matrix.android.sdk.api.session.crypto.crosssigning.SELF_SIGNING_KEY_S
 import org.matrix.android.sdk.api.session.crypto.crosssigning.USER_SIGNING_KEY_SSSS_NAME
 import org.matrix.android.sdk.api.session.crypto.keysbackup.KeysBackupState
 import org.matrix.android.sdk.api.session.crypto.keysbackup.KeysBackupStateListener
-import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 
 data class SignoutCheckViewState(
@@ -98,7 +98,9 @@ class SignoutCheckViewModel @AssistedInject constructor(
             )
         }
 
-        session.flow().liveUserAccountData(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
+        session.accountDataService()
+                .getLiveUserAccountDataEvents(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
+                .asFlow()
                 .map {
                     session.sharedSecretStorageService.isRecoverySetup()
                 }