Create extensions backupStateFlow() and recoveryStateFlow.

libraries/matrix/impl/src/main/kotlin/io/element/android/libraries/matrix/impl/encryption/EncryptionExtension.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2024 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 io.element.android.libraries.matrix.impl.encryption
+
+import io.element.android.libraries.matrix.api.encryption.BackupState
+import io.element.android.libraries.matrix.api.encryption.RecoveryState
+import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
+import kotlinx.coroutines.flow.Flow
+import org.matrix.rustcomponents.sdk.BackupStateListener
+import org.matrix.rustcomponents.sdk.EncryptionInterface
+import org.matrix.rustcomponents.sdk.RecoveryStateListener
+import org.matrix.rustcomponents.sdk.BackupState as RustBackupState
+import org.matrix.rustcomponents.sdk.RecoveryState as RustRecoveryState
+
+internal fun EncryptionInterface.backupStateFlow(): Flow<BackupState> = mxCallbackFlow {
+    val backupStateMapper = BackupStateMapper()
+    trySend(backupStateMapper.map(backupState()))
+    val listener = object : BackupStateListener {
+        override fun onUpdate(status: RustBackupState) {
+            trySend(backupStateMapper.map(status))
+        }
+    }
+    backupStateListener(listener)
+}
+
+internal fun EncryptionInterface.recoveryStateFlow(): Flow<RecoveryState> = mxCallbackFlow {
+    val recoveryStateMapper = RecoveryStateMapper()
+    trySend(recoveryStateMapper.map(recoveryState()))
+    val listener = object : RecoveryStateListener {
+        override fun onUpdate(status: RustRecoveryState) {
+            trySend(recoveryStateMapper.map(status))
+        }
+    }
+    recoveryStateListener(listener)
+}

libraries/matrix/impl/src/main/kotlin/io/element/android/libraries/matrix/impl/encryption/RustEncryptionService.kt

@@ -25,7 +25,6 @@ import io.element.android.libraries.matrix.api.encryption.EncryptionService
 import io.element.android.libraries.matrix.api.encryption.RecoveryState
 import io.element.android.libraries.matrix.api.sync.SyncState
 import io.element.android.libraries.matrix.impl.sync.RustSyncService
-import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.currentCoroutineContext
@@ -40,16 +39,12 @@ import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.withContext
-import org.matrix.rustcomponents.sdk.BackupStateListener
 import org.matrix.rustcomponents.sdk.BackupSteadyStateListener
 import org.matrix.rustcomponents.sdk.Client
 import org.matrix.rustcomponents.sdk.EnableRecoveryProgressListener
 import org.matrix.rustcomponents.sdk.Encryption
-import org.matrix.rustcomponents.sdk.RecoveryStateListener
-import org.matrix.rustcomponents.sdk.BackupState as RustBackupState
 import org.matrix.rustcomponents.sdk.BackupUploadState as RustBackupUploadState
 import org.matrix.rustcomponents.sdk.EnableRecoveryProgress as RustEnableRecoveryProgress
-import org.matrix.rustcomponents.sdk.RecoveryState as RustRecoveryState
 import org.matrix.rustcomponents.sdk.SteadyStateException as RustSteadyStateException
 
 internal class RustEncryptionService(
@@ -60,23 +55,12 @@ internal class RustEncryptionService(
 ) : EncryptionService {
     private val service: Encryption = client.encryption()
 
-    private val backupStateMapper = BackupStateMapper()
-    private val recoveryStateMapper = RecoveryStateMapper()
     private val enableRecoveryProgressMapper = EnableRecoveryProgressMapper()
     private val backupUploadStateMapper = BackupUploadStateMapper()
     private val steadyStateExceptionMapper = SteadyStateExceptionMapper()
 
-    private val backupStateFlow = mxCallbackFlow {
-        val listener = object : BackupStateListener {
-            override fun onUpdate(status: RustBackupState) {
-                trySend(backupStateMapper.map(status))
-            }
-        }
-        service.backupStateListener(listener)
-    }.stateIn(sessionCoroutineScope, SharingStarted.Eagerly, service.backupState().let(backupStateMapper::map))
-
     override val backupStateStateFlow = combine(
-        backupStateFlow,
+        service.backupStateFlow(),
         syncService.syncState,
     ) { backupState, syncState ->
         if (syncState == SyncState.Running) {
@@ -86,17 +70,8 @@ internal class RustEncryptionService(
         }
     }.stateIn(sessionCoroutineScope, SharingStarted.Eagerly, BackupState.WAITING_FOR_SYNC)
 
-    private val recoveryStateFlow = mxCallbackFlow {
-        val listener = object : RecoveryStateListener {
-            override fun onUpdate(status: RustRecoveryState) {
-                trySend(recoveryStateMapper.map(status))
-            }
-        }
-        service.recoveryStateListener(listener)
-    }.stateIn(sessionCoroutineScope, SharingStarted.Eagerly, service.recoveryState().let(recoveryStateMapper::map))
-
     override val recoveryStateStateFlow = combine(
-        recoveryStateFlow,
+        service.recoveryStateFlow(),
         syncService.syncState,
     ) { recoveryState, syncState ->
         if (syncState == SyncState.Running) {