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

This reverts commit d9b02a20d8.

matrix-sdk-android-flow/.gitignore

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

matrix-sdk-android-flow/build.gradle

@@ -0,0 +1,49 @@
+
+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

@@ -0,0 +1,21 @@
+# 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

@@ -0,0 +1,40 @@
+/*
+ * 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

@@ -0,0 +1,5 @@
+<?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

@@ -0,0 +1,83 @@
+/*
+ * 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

@@ -0,0 +1,138 @@
+/*
+ * 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/main/java/org/matrix/android/sdk/flow/OptionalFlow.kt

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

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

@@ -0,0 +1,33 @@
+/*
+ * 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,3 +5,4 @@ include ':diff-match-patch'
 include ':attachment-viewer'
 include ':multipicker'
 include ':library:ui-styles'
+include ':matrix-sdk-android-flow'

vector/build.gradle

@@ -323,6 +323,7 @@ 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,10 +16,16 @@
 
 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) {
@@ -32,5 +38,31 @@ 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/features/crypto/quads/SharedSecureStorageViewModel.kt

@@ -16,7 +16,6 @@
 
 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
@@ -44,7 +43,9 @@ 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
 
@@ -115,9 +116,8 @@ class SharedSecureStorageViewModel @AssistedInject constructor(
             }
         }
 
-        session.cryptoService()
+        session.flow()
-                .getLiveCryptoDeviceInfo(session.myUserId)
+                .liveUserCryptoDevices(session.myUserId)
-                .asFlow()
                 .distinctUntilChanged()
                 .execute {
                     copy(

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

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

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

@@ -15,7 +15,6 @@
  */
 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
@@ -37,6 +36,7 @@ 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,9 +85,8 @@ class DiscoverySettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeThreePids() {
-        session
+        session.flow()
-                .getThreePidsLive(true)
+                .liveThreePIds(true)
-                .asFlow()
                 .onEach {
                     retrieveBinding(it)
                 }

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

@@ -47,9 +47,12 @@ 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
@@ -101,10 +104,8 @@ class HomeActivityViewModel @AssistedInject constructor(
                 .crossSigningService().allPrivateKeysKnown()
 
         safeActiveSession
-                .cryptoService()
+                .flow()
-                .crossSigningService()
+                .liveCrossSigningInfo(safeActiveSession.myUserId)
-                .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,6 +48,7 @@ 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
@@ -95,13 +96,11 @@ class HomeDetailViewModel @AssistedInject constructor(@Assisted initialState: Ho
         updateShowDialPadTab()
         observeDataStore()
         callManager.addProtocolsCheckerListener(this)
-        session.getUserLive(session.myUserId)
+        session.flow().liveUser(session.myUserId).execute {
-                .asFlow()
+            copy(
-                .execute {
+                    myMatrixItem = it.invoke()?.getOrNull()?.toMatrixItem()
-                    copy(
+            )
-                            myMatrixItem = it.invoke()?.getOrNull()?.toMatrixItem()
+        }
-                    )
-                }
     }
 
     private fun observeDataStore() {
@@ -184,8 +183,8 @@ class HomeDetailViewModel @AssistedInject constructor(@Assisted initialState: Ho
     }
 
     private fun observeSyncState() {
-        session.getSyncStateLive()
+        session.flow()
-                .asFlow()
+                .liveSyncState()
                 .setOnEach { syncState ->
                     copy(syncState = syncState)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -42,6 +41,7 @@ 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.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow(),
+                session.flow().liveUserCryptoDevices(session.myUserId),
-                session.cryptoService().getLiveMyDevicesInfo().asFlow(),
+                session.flow().liveMyDevicesInfo(),
-                session.cryptoService().crossSigningService().getLiveCrossSigningPrivateKeys().asFlow()
+                session.flow().liveCrossSigningPrivateKeys()
         )
         { cryptoList, infoList, pInfo ->
             //                    Timber.v("## Detector trigger ${cryptoList.map { "${it.deviceId} ${it.trustLevel}" }}")
@@ -132,7 +132,7 @@ class UnknownDeviceDetectorSharedViewModel @AssistedInject constructor(@Assisted
                     )
                 }
 
-        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow()
+        session.flow().liveUserCryptoDevices(session.myUserId)
                 .distinctUntilChanged()
                 .sample(5_000)
                 .onEach {

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

@@ -16,7 +16,6 @@
 
 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
@@ -26,10 +25,13 @@ 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)
@@ -60,11 +62,11 @@ class BreadcrumbsViewModel @AssistedInject constructor(@Assisted initialState: B
     // PRIVATE METHODS *****************************************************************************
 
     private fun observeBreadcrumbs() {
-        session.getBreadcrumbsLive(roomSummaryQueryParams {
+        session.flow()
+                .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,7 +37,6 @@ 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
@@ -109,6 +108,8 @@ 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
@@ -253,12 +254,11 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeActiveRoomWidgets() {
-        session.widgetService()
+        session.flow()
-                .getRoomWidgetsLive(
+                .liveRoomWidgets(
                         roomId = initialState.roomId,
                         widgetId = QueryStringValue.NoCondition
                 )
-                .asFlow()
                 .map { widgets ->
                     widgets.filter { it.isActive }
                 }
@@ -287,9 +287,8 @@ class RoomDetailViewModel @AssistedInject constructor(
         val queryParams = roomMemberQueryParams {
             this.userId = QueryStringValue.Equals(session.myUserId, QueryStringValue.Case.SENSITIVE)
         }
-        room
+        room.flow()
-                .getRoomMembersLive(queryParams)
+                .liveRoomMembers(queryParams)
-                .asFlow()
                 .map {
                     it.firstOrNull().toOptional()
                 }
@@ -1506,8 +1505,8 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeSyncState() {
-        session.getSyncStateLive()
+        session.flow()
-                .asFlow()
+                .liveSyncState()
                 .setOnEach { syncState ->
                     copy(syncState = syncState)
                 }
@@ -1521,8 +1520,7 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeRoomSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(
@@ -1534,7 +1532,7 @@ class RoomDetailViewModel @AssistedInject constructor(
     private fun getUnreadState() {
         combine(
                 timelineEvents,
-                room.getRoomSummaryLive().asFlow().unwrap()
+                room.flow().liveRoomSummary().unwrap()
         ) { timelineEvents, roomSummary ->
             computeUnreadState(timelineEvents, roomSummary)
         }
@@ -1581,8 +1579,8 @@ class RoomDetailViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.getChangeMembershipsLive()
+        session.flow()
-                .asFlow()
+                .liveRoomChangeMembershipState()
                 .map {
                     it[initialState.roomId] ?: ChangeMembershipState.Unknown
                 }

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

@@ -15,7 +15,6 @@
  */
 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
@@ -29,7 +28,6 @@ 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
@@ -60,6 +58,8 @@ 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.getTimeLineEventLive(initialState.eventId)
+        room.flow()
-                .asFlow()
+                .liveTimelineEvent(initialState.eventId)
                 .unwrap()
                 .execute {
                     copy(timelineEvent = it)
@@ -149,8 +149,8 @@ class MessageActionsViewModel @AssistedInject constructor(@Assisted
         if (room == null) return
         eventIdFlow
                 .flatMapLatest { eventId ->
-                    room.getEventAnnotationsSummaryLive(eventId)
+                    room.flow()
-                            .asFlow()
+                            .liveAnnotationSummary(eventId)
                             .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,7 +17,6 @@
 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
@@ -43,6 +42,7 @@ import org.matrix.android.sdk.api.session.room.UpdatableLivePageResult
 import org.matrix.android.sdk.api.session.room.members.ChangeMembershipState
 import org.matrix.android.sdk.api.session.room.model.tag.RoomTag
 import org.matrix.android.sdk.api.session.room.state.isPublic
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 import javax.inject.Inject
 
@@ -95,8 +95,7 @@ class RoomListViewModel @Inject constructor(
                     )
                 }
 
-        session.getUserLive(session.myUserId)
+        session.flow().liveUser(session.myUserId)
-                .asFlow()
                 .map { it.getOrNull()?.getBestName() }
                 .distinctUntilChanged()
                 .execute {
@@ -107,8 +106,8 @@ class RoomListViewModel @Inject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.getChangeMembershipsLive()
+        session.flow()
-                .asFlow()
+                .liveRoomChangeMembershipState()
                 .setOnEach {
                     copy(roomMembershipChanges = it)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -36,6 +35,7 @@ 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,9 +82,10 @@ class InvitesAcceptor @Inject constructor(
         val roomQueryParams = roomSummaryQueryParams {
             this.memberships = listOf(Membership.INVITE)
         }
+        val flowSession = session.flow()
         combine(
-                session.getRoomSummariesLive(roomQueryParams).asFlow(),
+                flowSession.liveRoomSummaries(roomQueryParams),
-                session.getChangeMembershipsLive().asFlow().debounce(1000)
+                flowSession.liveRoomChangeMembershipState().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.asFlow
+import androidx.lifecycle.viewModelScope
 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.getUserLive(session.myUserId)
+        session.rx()
-                .asFlow()
+                .liveUser(session.myUserId)
                 .unwrap()
                 .map {
                     if (MatrixPatterns.isUserId(it.userId)) {

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

@@ -16,9 +16,6 @@
 
 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
@@ -27,13 +24,15 @@ 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
+        return room.flow()
-                .getStateEventLive(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+                .liveStateEvent(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,7 +16,6 @@
 
 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
@@ -28,7 +27,6 @@ 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
@@ -49,6 +47,8 @@ import org.matrix.android.sdk.api.session.room.Room
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.RoomSummary
 import org.matrix.android.sdk.api.util.Optional
+import org.matrix.android.sdk.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.getRoomSummaryLive()
+                    room.flow()
-                            .asFlow()
+                            .liveRoomSummary()
                             .unwrap()
                             .flowOn(Dispatchers.Default)
                             .map { mapToLeftViewEvent(room, it) }

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

@@ -16,7 +16,6 @@
 
 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
@@ -39,6 +38,7 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.roomdirectory.PublicRoomsFilter
 import org.matrix.android.sdk.api.session.room.model.roomdirectory.PublicRoomsParams
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 
 class RoomDirectoryViewModel @AssistedInject constructor(
@@ -80,8 +80,8 @@ class RoomDirectoryViewModel @AssistedInject constructor(
             memberships = listOf(Membership.JOIN)
         }
         session
-                .getRoomSummariesLive(queryParams)
+                .flow()
-                .asFlow()
+                .liveRoomSummaries(queryParams)
                 .map { roomSummaries ->
                     roomSummaries
                             .map { it.roomId }
@@ -93,8 +93,8 @@ class RoomDirectoryViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.getChangeMembershipsLive()
+        session.flow()
-                .asFlow()
+                .liveRoomChangeMembershipState()
                 .setOnEach {
                     copy(changeMembershipStates = it)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -43,6 +42,8 @@ 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,
@@ -167,8 +168,8 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
             excludeType = null
         }
         session
-                .getRoomSummariesLive(queryParams)
+                .flow()
-                .asFlow()
+                .liveRoomSummaries(queryParams)
                 .onEach { list ->
                     val isRoomJoined = list.any {
                         it.membership == Membership.JOIN
@@ -186,8 +187,8 @@ class RoomPreviewViewModel @AssistedInject constructor(@Assisted private val ini
     }
 
     private fun observeMembershipChanges() {
-        session.getChangeMembershipsLive()
+        session.flow()
-                .asFlow()
+                .liveRoomChangeMembershipState()
                 .onEach {
                     val changeMembership = it[initialState.roomId] ?: ChangeMembershipState.Unknown
                     val joinState = when (changeMembership) {

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

@@ -17,7 +17,6 @@
 
 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
@@ -32,7 +31,6 @@ 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
@@ -55,6 +53,8 @@ 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.cryptoService().getLiveCryptoDeviceInfo(initialState.userId).asFlow()
+        session.flow().liveUserCryptoDevices(initialState.userId)
                 .map {
                     Pair(
                             it.fold(true, { prev, dev -> prev && dev.isVerified }),
@@ -121,14 +121,14 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
                     )
                 }
 
-        session.cryptoService().crossSigningService().getLiveCrossSigningKeys(initialState.userId).asFlow()
+        session.flow().liveCrossSigningInfo(initialState.userId)
                 .execute {
                     copy(userMXCrossSigningInfo = it.invoke()?.getOrNull())
                 }
     }
 
     private fun observeIgnoredState() {
-        session.getIgnoredUsersLive().asFlow()
+        session.flow().liveIgnoredUsers()
                 .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.getRoomMembersLive(queryParams).asFlow()
+        room.flow().liveRoomMembers(queryParams)
                 .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.getRoomSummaryLive().asFlow().unwrap()
+        val roomSummaryLive = room.flow().liveRoomSummary().unwrap()
         val powerLevelsContentLive = PowerLevelsFlowFactory(room).createFlow()
 
         powerLevelsContentLive

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

@@ -16,7 +16,6 @@
  */
 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
@@ -34,7 +33,9 @@ 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,
@@ -55,17 +56,14 @@ class DeviceListBottomSheetViewModel @AssistedInject constructor(@Assisted priva
     }
 
     init {
-        session.cryptoService().getLiveCryptoDeviceInfo(args.userId)
+        session.flow().liveUserCryptoDevices(args.userId)
-                .asFlow()
                 .execute {
                     copy(cryptoDevices = it).also {
                         refreshSelectedId()
                     }
                 }
 
-        session.cryptoService().crossSigningService()
+        session.flow().liveCrossSigningInfo(args.userId)
-                .getLiveCrossSigningKeys(args.userId)
-                .asFlow()
                 .execute {
                     copy(memberCrossSigningKey = it.invoke()?.getOrNull())
                 }

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

@@ -17,7 +17,6 @@
 
 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
@@ -28,8 +27,6 @@ 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
@@ -43,6 +40,10 @@ import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.create.RoomCreateContent
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.api.session.room.state.isPublic
+import org.matrix.android.sdk.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,
@@ -68,14 +69,15 @@ class RoomProfileViewModel @AssistedInject constructor(
     private val room = session.getRoom(initialState.roomId)!!
 
     init {
-        observeRoomSummary()
+        val flowRoom = room.flow()
-        observeRoomCreateContent()
+        observeRoomSummary(flowRoom)
-        observeBannedRoomMembers()
+        observeRoomCreateContent(flowRoom)
+        observeBannedRoomMembers(flowRoom)
         observePermissions()
     }
 
-    private fun observeRoomCreateContent() {
+    private fun observeRoomCreateContent(flowRoom: FlowRoom) {
-        room.getStateEventLive(EventType.STATE_ROOM_CREATE, QueryStringValue.NoCondition).asFlow()
+        flowRoom.liveStateEvent(EventType.STATE_ROOM_CREATE, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomCreateContent>() }
                 .unwrap()
                 .execute { async ->
@@ -90,16 +92,16 @@ class RoomProfileViewModel @AssistedInject constructor(
                 }
     }
 
-    private fun observeRoomSummary() {
+    private fun observeRoomSummary(flowRoom: FlowRoom) {
-        room.getRoomSummaryLive().asFlow()
+        flowRoom.liveRoomSummary()
                 .unwrap()
                 .execute {
                     copy(roomSummary = it)
                 }
     }
 
-    private fun observeBannedRoomMembers() {
+    private fun observeBannedRoomMembers(flowRoom: FlowRoom) {
-        room.getRoomMembersLive(roomMemberQueryParams { memberships = listOf(Membership.BAN) }).asFlow()
+        flowRoom.liveRoomMembers(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
                 .execute {
                     copy(bannedMembership = it)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -29,8 +28,6 @@ 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
@@ -42,6 +39,11 @@ 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)
@@ -129,8 +131,7 @@ class RoomAliasViewModel @AssistedInject constructor(@Assisted initialState: Roo
     }
 
     private fun observeRoomSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(
@@ -172,8 +173,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.getStateEventLive(EventType.STATE_ROOM_CANONICAL_ALIAS, QueryStringValue.NoCondition)
+        room.flow()
-                .asFlow()
+                .liveStateEvent(EventType.STATE_ROOM_CANONICAL_ALIAS, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomCanonicalAliasContent>() }
                 .unwrap()
                 .setOnEach {

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

@@ -16,7 +16,6 @@
 
 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
@@ -27,7 +26,6 @@ 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
@@ -40,6 +38,10 @@ 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,
@@ -55,15 +57,13 @@ class RoomBannedMemberListViewModel @AssistedInject constructor(@Assisted initia
 
     init {
 
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)
                 }
 
-        room.getRoomMembersLive(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
+        room.flow().liveRoomMembers(roomMemberQueryParams { memberships = listOf(Membership.BAN) })
-                .asFlow()
                 .execute {
                     copy(
                             bannedMemberSummaries = it

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

@@ -27,8 +27,6 @@ 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
@@ -53,6 +51,9 @@ 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,
@@ -93,9 +94,9 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
         }
 
         combine(
-                room.getRoomMembersLive(roomMemberQueryParams).asFlow(),
+                room.flow().liveRoomMembers(roomMemberQueryParams),
-                room.getStateEventLive(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+                room.flow()
-                        .asFlow()
+                        .liveStateEvent(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
                         .mapOptional { it.content.toModel<PowerLevelsContent>() }
                         .unwrap()
         )
@@ -108,8 +109,7 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
                 }
 
         if (room.isEncrypted()) {
-            room.getRoomMembersLive(roomMemberQueryParams)
+            room.flow().liveRoomMembers(roomMemberQueryParams)
-                    .asFlow()
                     .flowOn(Dispatchers.Main)
                     .flatMapLatest { membersSummary ->
                         session.cryptoService().getLiveCryptoDeviceInfo(membersSummary.map { it.userId })
@@ -153,8 +153,7 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
     }
 
     private fun observeRoomSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)
@@ -162,8 +161,7 @@ class RoomMemberListViewModel @AssistedInject constructor(@Assisted initialState
     }
 
     private fun observeThirdPartyInvites() {
-        room.getStateEventsLive(setOf(EventType.STATE_ROOM_THIRD_PARTY_INVITE))
+        room.flow().liveStateEvents(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.asFlow
+import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.FragmentViewModelContext
 import com.airbnb.mvrx.MavericksViewModelFactory
 import com.airbnb.mvrx.Success
@@ -25,10 +25,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 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,
@@ -63,8 +66,7 @@ class RoomNotificationSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)
@@ -72,8 +74,8 @@ class RoomNotificationSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observeNotificationState() {
-        room.getLiveRoomNotificationState()
+        room.rx()
-                .asFlow()
+                .liveNotificationState()
                 .execute {
                     copy(notificationState = it)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -26,7 +25,6 @@ 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
@@ -36,6 +34,8 @@ import org.matrix.android.sdk.api.session.events.model.EventType
 import org.matrix.android.sdk.api.session.events.model.toContent
 import org.matrix.android.sdk.api.session.room.model.PowerLevelsContent
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 
 class RoomPermissionsViewModel @AssistedInject constructor(@Assisted initialState: RoomPermissionsViewState,
                                                            private val session: Session)
@@ -63,8 +63,7 @@ class RoomPermissionsViewModel @AssistedInject constructor(@Assisted initialStat
     }
 
     private fun observeRoomSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(

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

@@ -17,7 +17,6 @@
 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
@@ -27,8 +26,6 @@ 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
@@ -46,6 +43,9 @@ 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,8 +125,7 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeRoomSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     val roomSummary = async.invoke()
@@ -162,8 +161,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeRoomHistoryVisibility() {
-        room.getStateEventLive(EventType.STATE_ROOM_HISTORY_VISIBILITY, QueryStringValue.NoCondition)
+        room.flow()
-                .asFlow()
+                .liveStateEvent(EventType.STATE_ROOM_HISTORY_VISIBILITY, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomHistoryVisibilityContent>() }
                 .unwrap()
                 .mapNotNull { it.historyVisibility }
@@ -173,8 +172,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeJoinRule() {
-        room.getStateEventLive(EventType.STATE_ROOM_JOIN_RULES, QueryStringValue.NoCondition)
+        room.flow()
-                .asFlow()
+                .liveStateEvent(EventType.STATE_ROOM_JOIN_RULES, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomJoinRulesContent>() }
                 .unwrap()
                 .mapNotNull { it.joinRules }
@@ -184,8 +183,8 @@ class RoomSettingsViewModel @AssistedInject constructor(@Assisted initialState:
     }
 
     private fun observeGuestAccess() {
-        room.getStateEventLive(EventType.STATE_ROOM_GUEST_ACCESS, QueryStringValue.NoCondition)
+        room.flow()
-                .asFlow()
+                .liveStateEvent(EventType.STATE_ROOM_GUEST_ACCESS, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomGuestAccessContent>() }
                 .unwrap()
                 .mapNotNull { it.guestAccess }
@@ -198,8 +197,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.getStateEventLive(EventType.STATE_ROOM_AVATAR, QueryStringValue.NoCondition)
+        room.flow()
-                .asFlow()
+                .liveStateEvent(EventType.STATE_ROOM_AVATAR, QueryStringValue.NoCondition)
                 .mapOptional { it.content.toModel<RoomAvatarContent>() }
                 .unwrap()
                 .setOnEach {

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

@@ -16,7 +16,6 @@
 
 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
@@ -29,10 +28,11 @@ 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,8 +65,7 @@ class RoomUploadsViewModel @AssistedInject constructor(
     }
 
     private fun observeRoomSummary() {
-        room.getRoomSummaryLive()
+        room.flow().liveRoomSummary()
-                .asFlow()
                 .unwrap()
                 .execute { async ->
                     copy(roomSummary = async)

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

@@ -16,7 +16,6 @@
 
 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
@@ -26,6 +25,7 @@ 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,12 +39,11 @@ data class SecretsSynchronisationInfo(
 )
 
 fun Session.liveSecretSynchronisationInfo(): Flow<SecretsSynchronisationInfo> {
+    val sessionFlow = flow()
     return combine(
-            accountDataService()
+            sessionFlow.liveUserAccountData(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME, KEYBACKUP_SECRET_SSSS_NAME)),
-                    .getLiveUserAccountDataEvents(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME, KEYBACKUP_SECRET_SSSS_NAME))
+            sessionFlow.liveCrossSigningInfo(myUserId),
-                    .asFlow(),
+            sessionFlow.liveCrossSigningPrivateKeys()
-            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,7 +26,6 @@ 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
@@ -48,7 +47,6 @@ 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
@@ -64,6 +62,8 @@ 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.getUserLive(session.myUserId)
+        session.flow()
-                .asFlow()
+                .liveUser(session.myUserId)
                 .unwrap()
                 .distinctUntilChangedBy { user -> user.avatarUrl }
                 .onEach {
@@ -133,8 +133,8 @@ class VectorSettingsGeneralFragment @Inject constructor(
     }
 
     private fun observeUserDisplayName() {
-        session.getUserLive(session.myUserId)
+        session.flow()
-                .asFlow()
+                .liveUser(session.myUserId)
                 .unwrap()
                 .map { it.displayName ?: "" }
                 .distinctUntilChanged()

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

@@ -15,7 +15,6 @@
  */
 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
@@ -38,6 +37,7 @@ 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.cryptoService().getLiveMyDevicesInfo().asFlow(),
+                session.flow().liveMyDevicesInfo(),
-                session.cryptoService().crossSigningService().getLiveCrossSigningKeys(session.myUserId).asFlow()
+                session.flow().liveCrossSigningInfo(session.myUserId)
         )
         { myDevicesInfo, mxCrossSigningInfo ->
             myDevicesInfo to mxCrossSigningInfo

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

@@ -15,7 +15,6 @@
  */
 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
@@ -28,7 +27,9 @@ 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,
@@ -49,8 +50,7 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
                     isRecoverySetup = session.sharedSecretStorageService.isRecoverySetup()
             )
         }
-        session.cryptoService().crossSigningService().getLiveCrossSigningKeys(session.myUserId)
+        session.flow().liveCrossSigningInfo(session.myUserId)
-                .asFlow()
                 .execute {
                     copy(
                             hasAccountCrossSigning = it.invoke()?.getOrNull() != null,
@@ -58,8 +58,7 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
                     )
                 }
 
-        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId)
+        session.flow().liveUserCryptoDevices(session.myUserId)
-                .asFlow()
                 .map { list ->
                     list.firstOrNull { it.deviceId == deviceId }
                 }
@@ -70,7 +69,7 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
                     )
                 }
 
-        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow()
+        session.flow().liveUserCryptoDevices(session.myUserId)
                 .map { it.size }
                 .execute {
                     copy(
@@ -82,8 +81,7 @@ class DeviceVerificationInfoBottomSheetViewModel @AssistedInject constructor(@As
             copy(deviceInfo = Loading())
         }
 
-        session.cryptoService().getLiveMyDevicesInfo()
+        session.flow().liveMyDevicesInfo()
-                .asFlow()
                 .map { devices ->
                     devices.firstOrNull { it.deviceId == deviceId } ?: DeviceInfo(deviceId = deviceId)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -57,6 +56,7 @@ 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,14 +123,8 @@ class DevicesViewModel @AssistedInject constructor(
         }
 
         combine(
-                session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId).asFlow()
+                session.flow().liveUserCryptoDevices(session.myUserId),
-                        .onEach {
+                session.flow().liveMyDevicesInfo()
-                            Timber.v("getLiveCryptoDeviceInfo")
-                        },
-                session.cryptoService().getLiveMyDevicesInfo().asFlow()
-                        .onEach {
-                            Timber.v("getLiveMyDevicesInfo")
-                        }
         )
         { cryptoList, infoList ->
             infoList
@@ -147,8 +141,7 @@ class DevicesViewModel @AssistedInject constructor(
                     )
                 }
 
-        session.cryptoService().crossSigningService().getLiveCrossSigningKeys(session.myUserId)
+        session.flow().liveCrossSigningInfo(session.myUserId)
-                .asFlow()
                 .execute {
                     copy(
                             hasAccountCrossSigning = it.invoke()?.getOrNull() != null,
@@ -164,8 +157,7 @@ class DevicesViewModel @AssistedInject constructor(
 //                    )
 //                }
 
-        session.cryptoService().getLiveCryptoDeviceInfo(session.myUserId)
+        session.flow().liveUserCryptoDevices(session.myUserId)
-                .asFlow()
                 .map { it.size }
                 .distinctUntilChanged()
                 .sample(5_000)

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

@@ -16,7 +16,6 @@
 
 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
@@ -32,6 +31,7 @@ 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,9 +42,7 @@ class AccountDataViewModel @AssistedInject constructor(@Assisted initialState: A
     : VectorViewModel<AccountDataViewState, AccountDataAction, EmptyViewEvents>(initialState) {
 
     init {
-        session.accountDataService()
+        session.flow().liveUserAccountData(emptySet())
-                .getLiveUserAccountDataEvents(emptySet())
-                .asFlow()
                 .execute {
                     copy(accountData = it)
                 }

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

@@ -16,7 +16,6 @@
 
 package im.vector.app.features.settings.ignored
 
-import androidx.lifecycle.asFlow
 import com.airbnb.mvrx.Async
 import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.FragmentViewModelContext
@@ -34,6 +33,7 @@ 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.getIgnoredUsersLive()
+        session.flow()
-                .asFlow()
+                .liveIgnoredUsers()
                 .execute { async ->
                     copy(
                             ignoredUsers = async.invoke().orEmpty()

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

@@ -16,7 +16,6 @@
 
 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
@@ -39,6 +38,7 @@ 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.getThreePidsLive(true)
+        session.flow()
-                .asFlow()
+                .liveThreePIds(true)
                 .execute {
                     copy(
                             threePids = it
@@ -111,8 +111,8 @@ class ThreePidsSettingsViewModel @AssistedInject constructor(
     }
 
     private fun observePendingThreePids() {
-        session.getPendingThreePidsLive()
+        session.flow()
-                .asFlow()
+                .livePendingThreePIds()
                 .execute {
                     copy(
                             pendingThreePids = it,

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

@@ -16,7 +16,6 @@
 
 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
@@ -38,6 +37,7 @@ 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.getRoomSummariesLive(queryParams)
+        session
-                .asFlow()
+                .flow().liveRoomSummaries(queryParams)
                 .execute {
                     copy(roomSummaries = it)
                 }
@@ -86,7 +86,7 @@ class IncomingShareViewModel @AssistedInject constructor(
                         displayName = displayNameQuery
                         memberships = listOf(Membership.JOIN)
                     }
-                    session.getRoomSummariesLive(filterQueryParams).asFlow()
+                    session.flow().liveRoomSummaries(filterQueryParams)
                 }
                 .sample(300)
                 .map { it.sortedWith(breadcrumbsRoomComparator) }

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

@@ -16,7 +16,6 @@
 
 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
@@ -43,6 +42,8 @@ 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(
@@ -77,19 +78,17 @@ class SpaceMenuViewModel @AssistedInject constructor(
 
         session.getRoom(initialState.spaceId)?.let { room ->
 
-            room.getRoomSummaryLive()
+            room.flow().liveRoomSummary().onEach {
-                    .asFlow()
+                it.getOrNull()?.let {
-                    .onEach {
+                    if (it.membership == Membership.LEAVE) {
-                        it.getOrNull()?.let {
+                        setState { copy(leavingState = Success(Unit)) }
-                            if (it.membership == Membership.LEAVE) {
+                        if (appStateHandler.safeActiveSpaceId() == initialState.spaceId) {
-                                setState { copy(leavingState = Success(Unit)) }
+                            // switch to home?
-                                if (appStateHandler.safeActiveSpaceId() == initialState.spaceId) {
+                            appStateHandler.setCurrentSpace(null, session)
-                                    // 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,6 +53,7 @@ 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
 
@@ -81,13 +82,14 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
 
     init {
 
-        session.getUserLive(session.myUserId)
+        session.getUserLive(session.myUserId).asObservable()
-                .asFlow()
+                .subscribe {
-                .setOnEach {
+                    setState {
-                    copy(
+                        copy(
-                            myMxItem = it.getOrNull()?.toMatrixItem()?.let { Success(it) } ?: Loading()
+                                myMxItem = it?.getOrNull()?.toMatrixItem()?.let { Success(it) } ?: Loading()
-                    )
+                        )
-                }
+                    }
+                }.disposeOnClear()
 
         observeSpaceSummaries()
 //        observeSelectionState()
@@ -282,13 +284,16 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
                     null)
         }
 
+        val flowSession = session.flow()
+
         combine(
-                session.getUserLive(session.myUserId)
+                flowSession
-                        .asFlow()
+                        .liveUser(session.myUserId)
                         .map {
                             it.getOrNull()
                         },
-                session.spaceService().getSpaceSummariesLive(spaceSummaryQueryParams).asFlow(),
+                flowSession
+                        .liveSpaceSummaries(spaceSummaryQueryParams),
                 session
                         .accountDataService()
                         .getLiveRoomAccountDataEvents(setOf(RoomAccountDataTypes.EVENT_TYPE_SPACE_ORDER))
@@ -314,8 +319,7 @@ class SpacesListViewModel @AssistedInject constructor(@Assisted initialState: Sp
         // clear local echos on update
         session.accountDataService()
                 .getLiveRoomAccountDataEvents(setOf(RoomAccountDataTypes.EVENT_TYPE_SPACE_ORDER))
-                .asFlow()
+                .asObservable().execute {
-                .execute {
                     copy(
                             spaceOrderLocalEchos = emptyMap()
                     )

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

@@ -16,7 +16,6 @@
 
 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
@@ -44,6 +43,7 @@ import org.matrix.android.sdk.api.session.room.model.RoomType
 import org.matrix.android.sdk.api.session.room.model.SpaceChildInfo
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
 import org.matrix.android.sdk.api.session.room.roomSummaryQueryParams
+import org.matrix.android.sdk.flow.flow
 import timber.log.Timber
 
 class SpaceDirectoryViewModel @AssistedInject constructor(
@@ -147,8 +147,9 @@ class SpaceDirectoryViewModel @AssistedInject constructor(
             memberships = listOf(Membership.JOIN)
             excludeType = null
         }
-        session.getRoomSummariesLive(queryParams)
+        session
-                .asFlow()
+                .flow()
+                .liveRoomSummaries(queryParams)
                 .map {
                     it.map { it.roomId }.toSet()
                 }
@@ -158,8 +159,8 @@ class SpaceDirectoryViewModel @AssistedInject constructor(
     }
 
     private fun observeMembershipChanges() {
-        session.getChangeMembershipsLive()
+        session.flow()
-                .asFlow()
+                .liveRoomChangeMembershipState()
                 .setOnEach {
                     copy(changeMembershipStates = it)
                 }

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

@@ -16,7 +16,6 @@
 
 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
@@ -31,7 +30,6 @@ 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
@@ -41,6 +39,8 @@ 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,8 +97,7 @@ class SpaceLeaveAdvancedViewModel @AssistedInject constructor(
         val spaceSummary = session.getRoomSummary(initialState.spaceId)
         setState { copy(spaceSummary = spaceSummary) }
         session.getRoom(initialState.spaceId)?.let { room ->
-            room.getRoomSummaryLive()
+            room.flow().liveRoomSummary()
-                    .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,23 +29,21 @@ 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 kotlinx.coroutines.Dispatchers
+import io.reactivex.Single
-import kotlinx.coroutines.flow.MutableStateFlow
+import io.reactivex.android.schedulers.AndroidSchedulers
-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,
@@ -56,9 +54,9 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                                                     private val session: Session)
     : VectorViewModel<UserListViewState, UserListAction, UserListViewEvents>(initialState) {
 
-    private val knownUsersSearch = MutableStateFlow("")
+    private val knownUsersSearch = BehaviorRelay.create<KnownUsersSearch>()
-    private val directoryUsersSearch = MutableStateFlow("")
+    private val directoryUsersSearch = BehaviorRelay.create<DirectoryUsersSearch>()
-    private val identityServerUsersSearch = MutableStateFlow("")
+    private val identityServerUsersSearch = BehaviorRelay.create<String>()
 
     @AssistedFactory
     interface Factory {
@@ -79,10 +77,11 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private val identityServerListener = object : IdentityServiceListener {
         override fun onIdentityServerChange() {
             withState {
-                identityServerUsersSearch.tryEmit(it.searchTerm)
+                identityServerUsersSearch.accept(it.searchTerm)
-                val identityServerURL = cleanISURL(session.identityService().getCurrentIdentityServerUrl())
                 setState {
-                    copy(configuredIdentityServer = identityServerURL)
+                    copy(
+                            configuredIdentityServer = cleanISURL(session.identityService().getCurrentIdentityServerUrl())
+                    )
                 }
             }
         }
@@ -121,7 +120,7 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private fun handleISUpdateConsent(action: UserListAction.UpdateUserConsent) {
         session.identityService().setUserConsent(action.consent)
         withState {
-            identityServerUsersSearch.tryEmit(it.searchTerm)
+            identityServerUsersSearch.accept(it.searchTerm)
         }
     }
 
@@ -140,9 +139,9 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                 )
             }
         }
-        identityServerUsersSearch.tryEmit(searchTerm)
+        identityServerUsersSearch.accept(searchTerm)
-        knownUsersSearch.tryEmit(searchTerm)
+        knownUsersSearch.accept(searchTerm)
-        directoryUsersSearch.tryEmit(searchTerm)
+        directoryUsersSearch.accept(searchTerm)
     }
 
     private fun handleShareMyMatrixToLink() {
@@ -152,9 +151,9 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     }
 
     private fun handleClearSearchUsers() {
-        knownUsersSearch.tryEmit("")
+        knownUsersSearch.accept("")
-        directoryUsersSearch.tryEmit("")
+        directoryUsersSearch.accept("")
-        identityServerUsersSearch.tryEmit("")
+        identityServerUsersSearch.accept("")
         setState {
             copy(searchTerm = "")
         }
@@ -163,82 +162,103 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private fun observeUsers() = withState { state ->
         identityServerUsersSearch
                 .filter { it.isEmail() }
-                .sample(300)
+                .throttleLast(300, TimeUnit.MILLISECONDS)
-                .onEach { search ->
+                .switchMapSingle { search ->
-                    executeSearchEmail(search)
+                    val flowSession = session.rx()
-                }.launchIn(viewModelScope)
+                    val stream =
+                            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
-                .sample(300)
+                .throttleLast(300, TimeUnit.MILLISECONDS)
-                .flowOn(Dispatchers.Main)
+                .observeOn(AndroidSchedulers.mainThread())
-                .flatMapLatest { search ->
+                .switchMap {
-                    session.getPagedUsersLive(search, state.excludedUserIds).asFlow()
+                    session.rx().livePagedUsers(it, state.excludedUserIds)
-                }.execute {
+                }
-                    copy(knownUsers = it)
+                .execute { async ->
+                    copy(knownUsers = async)
                 }
 
         directoryUsersSearch
-                .debounce(300)
+                .debounce(300, TimeUnit.MILLISECONDS)
-                .onEach { search ->
+                .switchMapSingle { search ->
-                    executeSearchDirectory(state, search)
+                    val stream = if (search.isBlank()) {
-                }.launchIn(viewModelScope)
+                        Single.just(emptyList<User>())
-    }
+                    } 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()
+                                        )
+                                    }
 
-    private suspend fun executeSearchEmail(search: String) {
+                            Single.zip(
-        suspend {
+                                    searchObservable,
-            val params = listOf(ThreePid.Email(search))
+                                    profileObservable,
-            val foundThreePid = tryOrNull {
+                                    { searchResults, optionalProfile ->
-                session.identityService().lookUp(params).firstOrNull()
+                                        val profile = optionalProfile.getOrNull() ?: return@zip searchResults
-            }
+                                        val searchContainsProfile = searchResults.any { it.userId == profile.userId }
-            if (foundThreePid == null) {
+                                        if (searchContainsProfile) {
-                null
+                                            searchResults
-            } else {
+                                        } else {
-                try {
+                                            listOf(profile) + searchResults
-                    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) {
+                    }
-                    ThreePidUser(email = search, user = User(foundThreePid.matrixId))
+                    stream.toAsync {
+                        copy(directoryUsers = it)
+                    }
                 }
-            }
+                .subscribe()
-        }.execute {
+                .disposeOnClear()
-            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,7 +17,6 @@
 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
@@ -30,8 +29,6 @@ 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
@@ -45,6 +42,9 @@ 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,8 +119,7 @@ class WidgetViewModel @AssistedInject constructor(@Assisted val initialState: Wi
         if (room == null) {
             return
         }
-        room.getStateEventLive(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
+        room.flow().liveStateEvent(EventType.STATE_ROOM_POWER_LEVELS, QueryStringValue.NoCondition)
-                .asFlow()
                 .mapOptional { it.content.toModel<PowerLevelsContent>() }
                 .unwrap()
                 .map {
@@ -136,9 +135,8 @@ class WidgetViewModel @AssistedInject constructor(@Assisted val initialState: Wi
             return
         }
         val widgetId = initialState.widgetId ?: return
-        session.widgetService()
+        session.flow()
-                .getRoomWidgetsLive(initialState.roomId, QueryStringValue.Equals(widgetId))
+                .liveRoomWidgets(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,7 +15,6 @@
  */
 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
@@ -33,6 +32,7 @@ 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,9 +49,8 @@ class RoomWidgetPermissionViewModel @AssistedInject constructor(@Assisted val in
 
     private fun observeWidget() {
         val widgetId = initialState.widgetId ?: return
-        session.widgetService()
+        session.flow()
-                .getRoomWidgetsLive(initialState.roomId, QueryStringValue.Equals(widgetId))
+                .liveRoomWidgets(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,7 +17,6 @@
 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
@@ -42,6 +41,7 @@ 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,19 +92,9 @@ class ServerBackupStatusViewModel @AssistedInject constructor(@Assisted initialS
     init {
         session.cryptoService().keysBackupService().addListener(this)
         keysBackupState.value = session.cryptoService().keysBackupService().state
-        val liveUserAccountData = session.accountDataService()
+        val liveUserAccountData = session.flow().liveUserAccountData(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
-                .getLiveUserAccountDataEvents(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
+        val liveCrossSigningInfo = session.flow().liveCrossSigningInfo(session.myUserId)
-                .asFlow()
+        val liveCrossSigningPrivateKeys = session.flow().liveCrossSigningPrivateKeys()
-        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,7 +17,6 @@
 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
@@ -44,6 +43,7 @@ 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,9 +98,7 @@ class SignoutCheckViewModel @AssistedInject constructor(
             )
         }
 
-        session.accountDataService()
+        session.flow().liveUserAccountData(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
-                .getLiveUserAccountDataEvents(setOf(MASTER_KEY_SSSS_NAME, USER_SIGNING_KEY_SSSS_NAME, SELF_SIGNING_KEY_SSSS_NAME))
-                .asFlow()
                 .map {
                     session.sharedSecretStorageService.isRecoverySetup()
                 }