Mavericks 2: introduce startWith (like startWithCallable from matrix-android-sdk-rx)

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

@@ -0,0 +1,32 @@
+/*
+ * 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 kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.withContext
+
+internal fun <T> Flow<T>.startWith(supplier: suspend () -> T): Flow<T> {
+    return this
+            .onStart {
+                val value = withContext(Dispatchers.IO) {
+                    supplier()
+                }
+                emit(value)
+            }
+}

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

@@ -30,31 +30,50 @@ import org.matrix.android.sdk.api.session.room.notification.RoomNotificationStat
 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
+import org.matrix.android.sdk.api.util.toOptional
 
 class FlowRoom(private val room: Room) {
 
     fun liveRoomSummary(): Flow<Optional<RoomSummary>> {
         return room.getRoomSummaryLive().asFlow()
+                .startWith { 
+                    room.roomSummary().toOptional() 
+                }
     }
 
     fun liveRoomMembers(queryParams: RoomMemberQueryParams): Flow<List<RoomMemberSummary>> {
         return room.getRoomMembersLive(queryParams).asFlow()
+                .startWith {
+                    room.getRoomMembers(queryParams)
+                }
     }
 
     fun liveAnnotationSummary(eventId: String): Flow<Optional<EventAnnotationsSummary>> {
         return room.getEventAnnotationsSummaryLive(eventId).asFlow()
+                .startWith {
+                    room.getEventAnnotationsSummary(eventId).toOptional()
+                }
     }
 
     fun liveTimelineEvent(eventId: String): Flow<Optional<TimelineEvent>> {
         return room.getTimeLineEventLive(eventId).asFlow()
+                .startWith {
+                    room.getTimeLineEvent(eventId).toOptional()
+                }
     }
 
     fun liveStateEvent(eventType: String, stateKey: QueryStringValue): Flow<Optional<Event>> {
         return room.getStateEventLive(eventType, stateKey).asFlow()
+                .startWith {
+                    room.getStateEvent(eventType, stateKey).toOptional()
+                }
     }
 
     fun liveStateEvents(eventTypes: Set<String>): Flow<List<Event>> {
         return room.getStateEventsLive(eventTypes).asFlow()
+                .startWith {
+                    room.getStateEvents(eventTypes)
+                }
     }
 
     fun liveReadMarker(): Flow<Optional<String>> {
@@ -71,6 +90,9 @@ class FlowRoom(private val room: Room) {
 
     fun liveDraft(): Flow<Optional<UserDraft>> {
         return room.getDraftLive().asFlow()
+                .startWith {
+                    room.getDraft().toOptional()
+                }
     }
 
     fun liveNotificationState(): Flow<RoomNotificationState> {

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

@@ -37,6 +37,7 @@ 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.api.util.toOptional
 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
@@ -45,22 +46,37 @@ class RxFlow(private val session: Session) {
 
     fun liveRoomSummaries(queryParams: RoomSummaryQueryParams): Flow<List<RoomSummary>> {
         return session.getRoomSummariesLive(queryParams).asFlow()
+                .startWith {
+                    session.getRoomSummaries(queryParams)
+                }
     }
 
     fun liveGroupSummaries(queryParams: GroupSummaryQueryParams): Flow<List<GroupSummary>> {
         return session.getGroupSummariesLive(queryParams).asFlow()
+                .startWith {
+                    session.getGroupSummaries(queryParams)
+                }
     }
 
     fun liveSpaceSummaries(queryParams: SpaceSummaryQueryParams): Flow<List<RoomSummary>> {
         return session.spaceService().getSpaceSummariesLive(queryParams).asFlow()
+                .startWith {
+                    session.spaceService().getSpaceSummaries(queryParams)
+                }
     }
 
     fun liveBreadcrumbs(queryParams: RoomSummaryQueryParams): Flow<List<RoomSummary>> {
         return session.getBreadcrumbsLive(queryParams).asFlow()
+                .startWith {
+                    session.getBreadcrumbs(queryParams)
+                }
     }
 
     fun liveMyDevicesInfo(): Flow<List<DeviceInfo>> {
         return session.cryptoService().getLiveMyDevicesInfo().asFlow()
+                .startWith {
+                    session.cryptoService().getMyDevicesInfo()
+                }
     }
 
     fun liveSyncState(): Flow<SyncState> {
@@ -73,10 +89,16 @@ class RxFlow(private val session: Session) {
 
     fun liveUser(userId: String): Flow<Optional<User>> {
         return session.getUserLive(userId).asFlow()
+                .startWith {
+                    session.getUser(userId).toOptional()
+                }
     }
 
     fun liveRoomMember(userId: String, roomId: String): Flow<Optional<RoomMemberSummary>> {
         return session.getRoomMemberLive(userId, roomId).asFlow()
+                .startWith {
+                    session.getRoomMember(userId, roomId).toOptional()
+                }
     }
 
     fun liveUsers(): Flow<List<User>> {
@@ -93,30 +115,47 @@ class RxFlow(private val session: Session) {
 
     fun liveThreePIds(refreshData: Boolean): Flow<List<ThreePid>> {
         return session.getThreePidsLive(refreshData).asFlow()
+                .startWith { session.getThreePids() }
     }
 
     fun livePendingThreePIds(): Flow<List<ThreePid>> {
         return session.getPendingThreePidsLive().asFlow()
+                .startWith { session.getPendingThreePids() }
     }
 
     fun liveUserCryptoDevices(userId: String): Flow<List<CryptoDeviceInfo>> {
         return session.cryptoService().getLiveCryptoDeviceInfo(userId).asFlow()
+                .startWith {
+                    session.cryptoService().getCryptoDeviceInfo(userId)
+                }
     }
 
     fun liveCrossSigningInfo(userId: String): Flow<Optional<MXCrossSigningInfo>> {
         return session.cryptoService().crossSigningService().getLiveCrossSigningKeys(userId).asFlow()
+                .startWith {
+                    session.cryptoService().crossSigningService().getUserCrossSigningKeys(userId).toOptional()
+                }
     }
 
     fun liveCrossSigningPrivateKeys(): Flow<Optional<PrivateKeysInfo>> {
         return session.cryptoService().crossSigningService().getLiveCrossSigningPrivateKeys().asFlow()
+                .startWith {
+                    session.cryptoService().crossSigningService().getCrossSigningPrivateKeys().toOptional()
+                }
     }
 
     fun liveUserAccountData(types: Set<String>): Flow<List<UserAccountDataEvent>> {
         return session.accountDataService().getLiveUserAccountDataEvents(types).asFlow()
+                .startWith {
+                    session.accountDataService().getUserAccountDataEvents(types)
+                }
     }
 
     fun liveRoomAccountData(types: Set<String>): Flow<List<RoomAccountDataEvent>> {
         return session.accountDataService().getLiveRoomAccountDataEvents(types).asFlow()
+                .startWith {
+                    session.accountDataService().getRoomAccountDataEvents(types)
+                }
     }
 
     fun liveRoomWidgets(
@@ -126,6 +165,9 @@ class RxFlow(private val session: Session) {
             excludedTypes: Set<String>? = null
     ): Flow<List<Widget>> {
         return session.widgetService().getRoomWidgetsLive(roomId, widgetId, widgetTypes, excludedTypes).asFlow()
+                .startWith {
+                    session.widgetService().getRoomWidgets(roomId, widgetId, widgetTypes, excludedTypes)
+                }
     }
 
     fun liveRoomChangeMembershipState(): Flow<Map<String, ChangeMembershipState>> {