Merge pull request #3886 from element-hq/feature/bma/fixSendQueueCrash

fix : protect some usages of client to avoid crashes
This commit is contained in:
ganfra 2024-11-22 13:11:45 +01:00 committed by GitHub
commit 20674dd535
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 156 additions and 111 deletions

View file

@ -28,6 +28,7 @@ import io.element.android.libraries.matrix.api.MatrixClient
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.roomlist.RoomListService
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.verification.SessionVerificationService
import io.element.android.libraries.matrix.api.verification.SessionVerifiedStatus
import io.element.android.libraries.preferences.api.store.EnableNativeSlidingSyncUseCase
@ -102,10 +103,7 @@ class LoggedInPresenter @Inject constructor(
}
}
LoggedInEvents.CheckSlidingSyncProxyAvailability -> coroutineScope.launch {
// Force the user to log out if they were using the proxy sliding sync and it's no longer available, but native sliding sync is.
forceNativeSlidingSyncMigration = !matrixClient.isUsingNativeSlidingSync() &&
matrixClient.isNativeSlidingSyncSupported() &&
!matrixClient.isSlidingSyncProxySupported()
forceNativeSlidingSyncMigration = matrixClient.forceNativeSlidingSyncMigration().getOrDefault(false)
}
LoggedInEvents.LogoutAndMigrateToNativeSlidingSync -> coroutineScope.launch {
// Enable native sliding sync if it wasn't already the case
@ -125,6 +123,18 @@ class LoggedInPresenter @Inject constructor(
)
}
// Force the user to log out if they were using the proxy sliding sync and it's no longer available, but native sliding sync is.
private suspend fun MatrixClient.forceNativeSlidingSyncMigration(): Result<Boolean> = runCatching {
val currentSlidingSyncVersion = currentSlidingSyncVersion().getOrThrow()
if (currentSlidingSyncVersion == SlidingSyncVersion.Proxy) {
val availableSlidingSyncVersions = availableSlidingSyncVersions().getOrThrow()
availableSlidingSyncVersions.contains(SlidingSyncVersion.Native) &&
!availableSlidingSyncVersions.contains(SlidingSyncVersion.Proxy)
} else {
false
}
}
private suspend fun ensurePusherIsRegistered(pusherRegistrationState: MutableState<AsyncData<Unit>>) {
Timber.tag(pusherTag.value).d("Ensure pusher is registered")
val currentPushProvider = pushService.getCurrentPushProvider()

View file

@ -21,6 +21,7 @@ import io.element.android.libraries.matrix.api.core.SessionId
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.roomlist.RoomListService
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.verification.SessionVerificationService
import io.element.android.libraries.matrix.api.verification.SessionVerifiedStatus
import io.element.android.libraries.matrix.test.AN_EXCEPTION
@ -501,9 +502,8 @@ class LoggedInPresenterTest {
// - The sliding sync proxy is no longer supported
// - The native sliding sync is supported
val matrixClient = FakeMatrixClient(
isUsingNativeSlidingSyncLambda = { false },
isSlidingSyncProxySupportedLambda = { false },
isNativeSlidingSyncSupportedLambda = { true },
currentSlidingSyncVersionLambda = { Result.success(SlidingSyncVersion.Proxy) },
availableSlidingSyncVersionsLambda = { Result.success(listOf(SlidingSyncVersion.Native)) },
)
val presenter = createLoggedInPresenter(matrixClient = matrixClient)
moleculeFlow(RecompositionMode.Immediate) {
@ -521,9 +521,8 @@ class LoggedInPresenterTest {
@Test
fun `present - CheckSlidingSyncProxyAvailability will not force the migration if native sliding sync is not supported too`() = runTest {
val matrixClient = FakeMatrixClient(
isUsingNativeSlidingSyncLambda = { false },
isSlidingSyncProxySupportedLambda = { false },
isNativeSlidingSyncSupportedLambda = { false },
currentSlidingSyncVersionLambda = { Result.success(SlidingSyncVersion.Proxy) },
availableSlidingSyncVersionsLambda = { Result.success(emptyList()) },
)
val presenter = createLoggedInPresenter(matrixClient = matrixClient)
moleculeFlow(RecompositionMode.Immediate) {

View file

@ -39,7 +39,6 @@ import io.element.android.features.roomlist.impl.search.RoomListSearchEvents
import io.element.android.features.roomlist.impl.search.RoomListSearchState
import io.element.android.libraries.architecture.AsyncData
import io.element.android.libraries.architecture.Presenter
import io.element.android.libraries.core.bool.orFalse
import io.element.android.libraries.designsystem.utils.snackbar.SnackbarDispatcher
import io.element.android.libraries.designsystem.utils.snackbar.collectSnackbarMessageAsState
import io.element.android.libraries.featureflag.api.FeatureFlagService
@ -51,6 +50,7 @@ import io.element.android.libraries.matrix.api.core.RoomId
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.roomlist.RoomList
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.timeline.ReceiptType
import io.element.android.libraries.preferences.api.store.SessionPreferencesStore
import io.element.android.libraries.push.api.notifications.NotificationCleaner
@ -231,10 +231,7 @@ class RoomListPresenter @Inject constructor(
}
}
val needsSlidingSyncMigration by produceState(false) {
value = runCatching {
// Note: this can fail when the session is destroyed from another client.
client.isNativeSlidingSyncSupported() && !client.isUsingNativeSlidingSync()
}.getOrNull().orFalse()
value = client.needsSlidingSyncMigration().getOrDefault(false)
}
val securityBannerState by rememberSecurityBannerState(securityBannerDismissed, needsSlidingSyncMigration)
return when {
@ -315,6 +312,19 @@ class RoomListPresenter @Inject constructor(
}
}
/**
* Checks if the user needs to migrate to a native sliding sync version.
*/
private suspend fun MatrixClient.needsSlidingSyncMigration(): Result<Boolean> = runCatching {
val currentSlidingSyncVersion = currentSlidingSyncVersion().getOrThrow()
if (currentSlidingSyncVersion != SlidingSyncVersion.Native) {
val availableSlidingSyncVersions = availableSlidingSyncVersions().getOrThrow()
availableSlidingSyncVersions.contains(SlidingSyncVersion.Native)
} else {
false
}
}
private var currentUpdateVisibleRangeJob: Job? = null
private fun CoroutineScope.updateVisibleRange(range: IntRange) {
currentUpdateVisibleRangeJob?.cancel()

View file

@ -30,6 +30,7 @@ import io.element.android.libraries.matrix.api.room.preview.RoomPreviewInfo
import io.element.android.libraries.matrix.api.roomdirectory.RoomDirectoryService
import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.roomlist.RoomSummary
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.sync.SyncService
import io.element.android.libraries.matrix.api.user.MatrixSearchUserResults
import io.element.android.libraries.matrix.api.user.MatrixUser
@ -145,14 +146,15 @@ interface MatrixClient : Closeable {
suspend fun getUrl(url: String): Result<String>
suspend fun getRoomPreviewInfo(roomIdOrAlias: RoomIdOrAlias, serverNames: List<String>): Result<RoomPreviewInfo>
/** Returns `true` if the home server supports native sliding sync. */
suspend fun isNativeSlidingSyncSupported(): Boolean
/**
* Returns the currently used sliding sync version.
*/
suspend fun currentSlidingSyncVersion(): Result<SlidingSyncVersion>
/** Returns `true` if the home server supports sliding sync using a proxy. */
suspend fun isSlidingSyncProxySupported(): Boolean
/** Returns `true` if the current session is using native sliding sync, `false` if it's using a proxy. */
fun isUsingNativeSlidingSync(): Boolean
/**
* Returns the available sliding sync versions for the current user.
*/
suspend fun availableSlidingSyncVersions(): Result<List<SlidingSyncVersion>>
fun canDeactivateAccount(): Boolean
suspend fun deactivateAccount(password: String, eraseData: Boolean): Result<Unit>

View file

@ -0,0 +1,14 @@
/*
* Copyright 2024 New Vector Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only
* Please see LICENSE in the repository root for full details.
*/
package io.element.android.libraries.matrix.api.sync
sealed interface SlidingSyncVersion {
data object None : SlidingSyncVersion
data object Proxy : SlidingSyncVersion
data object Native : SlidingSyncVersion
}

View file

@ -12,6 +12,7 @@ import io.element.android.libraries.androidutils.file.safeDelete
import io.element.android.libraries.core.bool.orFalse
import io.element.android.libraries.core.coroutine.CoroutineDispatchers
import io.element.android.libraries.core.coroutine.childScope
import io.element.android.libraries.core.data.tryOrNull
import io.element.android.libraries.featureflag.api.FeatureFlagService
import io.element.android.libraries.matrix.api.MatrixClient
import io.element.android.libraries.matrix.api.core.DeviceId
@ -41,6 +42,7 @@ import io.element.android.libraries.matrix.api.room.preview.RoomPreviewInfo
import io.element.android.libraries.matrix.api.roomdirectory.RoomDirectoryService
import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.roomlist.RoomSummary
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.sync.SyncService
import io.element.android.libraries.matrix.api.sync.SyncState
import io.element.android.libraries.matrix.api.user.MatrixSearchUserResults
@ -62,6 +64,7 @@ import io.element.android.libraries.matrix.impl.roomdirectory.RustRoomDirectoryS
import io.element.android.libraries.matrix.impl.roomlist.RoomListFactory
import io.element.android.libraries.matrix.impl.roomlist.RustRoomListService
import io.element.android.libraries.matrix.impl.sync.RustSyncService
import io.element.android.libraries.matrix.impl.sync.map
import io.element.android.libraries.matrix.impl.usersearch.UserProfileMapper
import io.element.android.libraries.matrix.impl.usersearch.UserSearchResultMapper
import io.element.android.libraries.matrix.impl.util.SessionPathsProvider
@ -100,7 +103,6 @@ import org.matrix.rustcomponents.sdk.IgnoredUsersListener
import org.matrix.rustcomponents.sdk.NotificationProcessSetup
import org.matrix.rustcomponents.sdk.PowerLevels
import org.matrix.rustcomponents.sdk.SendQueueRoomErrorListener
import org.matrix.rustcomponents.sdk.SlidingSyncVersion
import org.matrix.rustcomponents.sdk.TaskHandle
import org.matrix.rustcomponents.sdk.use
import timber.log.Timber
@ -116,44 +118,44 @@ import org.matrix.rustcomponents.sdk.RoomVisibility as RustRoomVisibility
import org.matrix.rustcomponents.sdk.SyncService as ClientSyncService
class RustMatrixClient(
private val client: Client,
private val innerClient: Client,
private val baseDirectory: File,
private val sessionStore: SessionStore,
private val appCoroutineScope: CoroutineScope,
private val sessionDelegate: RustClientSessionDelegate,
syncService: ClientSyncService,
innerSyncService: ClientSyncService,
dispatchers: CoroutineDispatchers,
baseCacheDirectory: File,
clock: SystemClock,
timelineEventTypeFilterFactory: TimelineEventTypeFilterFactory,
featureFlagService: FeatureFlagService,
) : MatrixClient {
override val sessionId: UserId = UserId(client.userId())
override val deviceId: DeviceId = DeviceId(client.deviceId())
override val sessionId: UserId = UserId(innerClient.userId())
override val deviceId: DeviceId = DeviceId(innerClient.deviceId())
override val sessionCoroutineScope = appCoroutineScope.childScope(dispatchers.main, "Session-$sessionId")
private val innerRoomListService = syncService.roomListService()
private val sessionDispatcher = dispatchers.io.limitedParallelism(64)
private val rustSyncService = RustSyncService(syncService, sessionCoroutineScope)
private val innerRoomListService = innerSyncService.roomListService()
private val rustSyncService = RustSyncService(innerSyncService, sessionCoroutineScope)
private val pushersService = RustPushersService(
client = client,
client = innerClient,
dispatchers = dispatchers,
)
private val notificationProcessSetup = NotificationProcessSetup.SingleProcess(syncService)
private val notificationClient = runBlocking { client.notificationClient(notificationProcessSetup) }
private val notificationService = RustNotificationService(notificationClient, dispatchers, clock)
private val notificationSettingsService = RustNotificationSettingsService(client, dispatchers)
private val notificationProcessSetup = NotificationProcessSetup.SingleProcess(innerSyncService)
private val innerNotificationClient = runBlocking { innerClient.notificationClient(notificationProcessSetup) }
private val notificationService = RustNotificationService(innerNotificationClient, dispatchers, clock)
private val notificationSettingsService = RustNotificationSettingsService(innerClient, dispatchers)
.apply { start() }
private val encryptionService = RustEncryptionService(
client = client,
client = innerClient,
syncService = rustSyncService,
sessionCoroutineScope = sessionCoroutineScope,
dispatchers = dispatchers,
)
private val roomDirectoryService = RustRoomDirectoryService(
client = client,
client = innerClient,
sessionDispatcher = sessionDispatcher,
)
@ -173,13 +175,12 @@ class RustMatrixClient(
)
private val verificationService = RustSessionVerificationService(
client = client,
client = innerClient,
isSyncServiceReady = rustSyncService.syncState.map { it == SyncState.Running },
sessionCoroutineScope = sessionCoroutineScope,
)
private val roomMembershipObserver = RoomMembershipObserver()
private val roomFactory = RustRoomFactory(
roomListService = roomListService,
innerRoomListService = innerRoomListService,
@ -199,24 +200,24 @@ class RustMatrixClient(
override val mediaLoader: MatrixMediaLoader = RustMediaLoader(
baseCacheDirectory = baseCacheDirectory,
dispatchers = dispatchers,
innerClient = client,
innerClient = innerClient,
)
private var clientDelegateTaskHandle: TaskHandle? = client.setDelegate(sessionDelegate)
private var clientDelegateTaskHandle: TaskHandle? = innerClient.setDelegate(sessionDelegate)
private val _userProfile: MutableStateFlow<MatrixUser> = MutableStateFlow(
MatrixUser(
userId = sessionId,
// TODO cache for displayName?
displayName = null,
avatarUrl = client.cachedAvatarUrl(),
avatarUrl = innerClient.cachedAvatarUrl(),
)
)
override val userProfile: StateFlow<MatrixUser> = _userProfile
override val ignoredUsersFlow = mxCallbackFlow<ImmutableList<UserId>> {
client.subscribeToIgnoredUsers(object : IgnoredUsersListener {
innerClient.subscribeToIgnoredUsers(object : IgnoredUsersListener {
override fun call(ignoredUserIds: List<String>) {
channel.trySend(ignoredUserIds.map(::UserId).toPersistentList())
}
@ -237,7 +238,7 @@ class RustMatrixClient(
override fun userIdServerName(): String {
return runCatching {
client.userIdServerName()
innerClient.userIdServerName()
}
.onFailure {
Timber.w(it, "Failed to get userIdServerName")
@ -248,7 +249,7 @@ class RustMatrixClient(
override suspend fun getUrl(url: String): Result<String> = withContext(sessionDispatcher) {
runCatching {
client.getUrl(url)
innerClient.getUrl(url)
}
}
@ -278,23 +279,23 @@ class RustMatrixClient(
.filter { roomSummary -> roomSummary.info.currentUserMembership == currentUserMembership }
.first()
// Ensure that the room is ready
.also { client.awaitRoomRemoteEcho(it.roomId.value) }
.also { innerClient.awaitRoomRemoteEcho(it.roomId.value) }
}
}
override suspend fun findDM(userId: UserId): RoomId? {
return client.getDmRoom(userId.value)?.use { RoomId(it.id()) }
return innerClient.getDmRoom(userId.value)?.use { RoomId(it.id()) }
}
override suspend fun ignoreUser(userId: UserId): Result<Unit> = withContext(sessionDispatcher) {
runCatching {
client.ignoreUser(userId.value)
innerClient.ignoreUser(userId.value)
}
}
override suspend fun unignoreUser(userId: UserId): Result<Unit> = withContext(sessionDispatcher) {
runCatching {
client.unignoreUser(userId.value)
innerClient.unignoreUser(userId.value)
}
}
@ -337,7 +338,7 @@ class RustMatrixClient(
},
canonicalAlias = createRoomParams.roomAliasName.getOrNull(),
)
val roomId = RoomId(client.createRoom(rustParams))
val roomId = RoomId(innerClient.createRoom(rustParams))
// Wait to receive the room back from the sync but do not returns failure if it fails.
try {
awaitRoom(roomId.toRoomIdOrAlias(), 30.seconds, CurrentUserMembership.JOINED)
@ -362,7 +363,7 @@ class RustMatrixClient(
override suspend fun getProfile(userId: UserId): Result<MatrixUser> = withContext(sessionDispatcher) {
runCatching {
client.getProfile(userId.value).let(UserProfileMapper::map)
innerClient.getProfile(userId.value).let(UserProfileMapper::map)
}
}
@ -372,28 +373,28 @@ class RustMatrixClient(
override suspend fun searchUsers(searchTerm: String, limit: Long): Result<MatrixSearchUserResults> =
withContext(sessionDispatcher) {
runCatching {
client.searchUsers(searchTerm, limit.toULong()).let(UserSearchResultMapper::map)
innerClient.searchUsers(searchTerm, limit.toULong()).let(UserSearchResultMapper::map)
}
}
override suspend fun setDisplayName(displayName: String): Result<Unit> =
withContext(sessionDispatcher) {
runCatching { client.setDisplayName(displayName) }
runCatching { innerClient.setDisplayName(displayName) }
}
override suspend fun uploadAvatar(mimeType: String, data: ByteArray): Result<Unit> =
withContext(sessionDispatcher) {
runCatching { client.uploadAvatar(mimeType, data) }
runCatching { innerClient.uploadAvatar(mimeType, data) }
}
override suspend fun removeAvatar(): Result<Unit> =
withContext(sessionDispatcher) {
runCatching { client.removeAvatar() }
runCatching { innerClient.removeAvatar() }
}
override suspend fun joinRoom(roomId: RoomId): Result<RoomSummary?> = withContext(sessionDispatcher) {
runCatching {
client.joinRoomById(roomId.value).destroy()
innerClient.joinRoomById(roomId.value).destroy()
try {
awaitRoom(roomId.toRoomIdOrAlias(), 10.seconds, CurrentUserMembership.JOINED)
} catch (e: Exception) {
@ -405,7 +406,7 @@ class RustMatrixClient(
override suspend fun joinRoomByIdOrAlias(roomIdOrAlias: RoomIdOrAlias, serverNames: List<String>): Result<RoomSummary?> = withContext(sessionDispatcher) {
runCatching {
client.joinRoomByIdOrAlias(
innerClient.joinRoomByIdOrAlias(
roomIdOrAlias = roomIdOrAlias.identifier,
serverNames = serverNames,
).destroy()
@ -422,7 +423,7 @@ class RustMatrixClient(
sessionDispatcher
) {
runCatching {
client.knock(roomIdOrAlias.identifier, message, serverNames).destroy()
innerClient.knock(roomIdOrAlias.identifier, message, serverNames).destroy()
try {
awaitRoom(roomIdOrAlias, 10.seconds, CurrentUserMembership.KNOCKED)
} catch (e: Exception) {
@ -434,19 +435,19 @@ class RustMatrixClient(
override suspend fun trackRecentlyVisitedRoom(roomId: RoomId): Result<Unit> = withContext(sessionDispatcher) {
runCatching {
client.trackRecentlyVisitedRoom(roomId.value)
innerClient.trackRecentlyVisitedRoom(roomId.value)
}
}
override suspend fun getRecentlyVisitedRooms(): Result<List<RoomId>> = withContext(sessionDispatcher) {
runCatching {
client.getRecentlyVisitedRooms().map(::RoomId)
innerClient.getRecentlyVisitedRooms().map(::RoomId)
}
}
override suspend fun resolveRoomAlias(roomAlias: RoomAlias): Result<Optional<ResolvedRoomAlias>> = withContext(sessionDispatcher) {
runCatching {
val result = client.resolveRoomAlias(roomAlias.value)?.let {
val result = innerClient.resolveRoomAlias(roomAlias.value)?.let {
ResolvedRoomAlias(
roomId = RoomId(it.roomId),
servers = it.servers,
@ -459,8 +460,8 @@ class RustMatrixClient(
override suspend fun getRoomPreviewInfo(roomIdOrAlias: RoomIdOrAlias, serverNames: List<String>): Result<RoomPreviewInfo> = withContext(sessionDispatcher) {
runCatching {
when (roomIdOrAlias) {
is RoomIdOrAlias.Alias -> client.getRoomPreviewFromRoomAlias(roomIdOrAlias.roomAlias.value)
is RoomIdOrAlias.Id -> client.getRoomPreviewFromRoomId(roomIdOrAlias.roomId.value, serverNames)
is RoomIdOrAlias.Alias -> innerClient.getRoomPreviewFromRoomAlias(roomIdOrAlias.roomAlias.value)
is RoomIdOrAlias.Id -> innerClient.getRoomPreviewFromRoomId(roomIdOrAlias.roomId.value, serverNames)
}.use { roomPreview ->
RoomPreviewInfoMapper.map(roomPreview.info())
}
@ -490,11 +491,6 @@ class RustMatrixClient(
clientDelegateTaskHandle?.cancelAndDestroy()
notificationSettingsService.destroy()
verificationService.destroy()
innerRoomListService.destroy()
notificationClient.destroy()
notificationProcessSetup.destroy()
encryptionService.destroy()
client.destroy()
}
override suspend fun getCacheSize(): Long {
@ -514,13 +510,13 @@ class RustMatrixClient(
withContext(sessionDispatcher) {
if (userInitiated) {
try {
result = client.logout()
result = innerClient.logout()
} catch (failure: Throwable) {
if (ignoreSdkError) {
Timber.e(failure, "Fail to call logout on HS. Still delete local files.")
} else {
// If the logout failed we need to restore the delegate
clientDelegateTaskHandle = client.setDelegate(sessionDelegate)
clientDelegateTaskHandle = innerClient.setDelegate(sessionDelegate)
Timber.e(failure, "Fail to call logout on HS.")
throw failure
}
@ -538,7 +534,7 @@ class RustMatrixClient(
override fun canDeactivateAccount(): Boolean {
return runCatching {
client.canDeactivateAccount()
innerClient.canDeactivateAccount()
}
.getOrNull()
.orFalse()
@ -552,7 +548,7 @@ class RustMatrixClient(
runCatching {
// First call without AuthData, should fail
val firstAttempt = runCatching {
client.deactivateAccount(
innerClient.deactivateAccount(
authData = null,
eraseData = eraseData,
)
@ -561,7 +557,7 @@ class RustMatrixClient(
Timber.w(firstAttempt.exceptionOrNull(), "Expected failure, try again")
// This is expected, try again with the password
runCatching {
client.deactivateAccount(
innerClient.deactivateAccount(
authData = AuthData.Password(
passwordDetails = AuthDataPasswordDetails(
identifier = sessionId.value,
@ -573,7 +569,7 @@ class RustMatrixClient(
}.onFailure {
Timber.e(it, "Failed to deactivate account")
// If the deactivation failed we need to restore the delegate
clientDelegateTaskHandle = client.setDelegate(sessionDelegate)
clientDelegateTaskHandle = innerClient.setDelegate(sessionDelegate)
throw it
}
}
@ -588,13 +584,13 @@ class RustMatrixClient(
override suspend fun getAccountManagementUrl(action: AccountManagementAction?): Result<String?> = withContext(sessionDispatcher) {
val rustAction = action?.toRustAction()
runCatching {
client.accountUrl(rustAction)
innerClient.accountUrl(rustAction)
}
}
override suspend fun uploadMedia(mimeType: String, data: ByteArray, progressCallback: ProgressCallback?): Result<String> = withContext(sessionDispatcher) {
runCatching {
client.uploadMedia(mimeType, data, progressCallback?.toProgressWatcher())
innerClient.uploadMedia(mimeType, data, progressCallback?.toProgressWatcher())
}
}
@ -617,29 +613,33 @@ class RustMatrixClient(
.distinctUntilChanged()
}
override suspend fun setAllSendQueuesEnabled(enabled: Boolean) = withContext(sessionDispatcher) {
Timber.i("setAllSendQueuesEnabled($enabled)")
client.enableAllSendQueues(enabled)
override suspend fun setAllSendQueuesEnabled(enabled: Boolean) {
withContext(sessionDispatcher) {
Timber.i("setAllSendQueuesEnabled($enabled)")
tryOrNull {
innerClient.enableAllSendQueues(enabled)
}
}
}
override fun sendQueueDisabledFlow(): Flow<RoomId> = mxCallbackFlow {
client.subscribeToSendQueueStatus(object : SendQueueRoomErrorListener {
innerClient.subscribeToSendQueueStatus(object : SendQueueRoomErrorListener {
override fun onError(roomId: String, error: ClientException) {
trySend(RoomId(roomId))
}
})
}.buffer(Channel.UNLIMITED)
override suspend fun isNativeSlidingSyncSupported(): Boolean {
return client.availableSlidingSyncVersions().contains(SlidingSyncVersion.Native)
override suspend fun availableSlidingSyncVersions(): Result<List<SlidingSyncVersion>> = withContext(sessionDispatcher) {
runCatching {
innerClient.availableSlidingSyncVersions().map { it.map() }
}
}
override suspend fun isSlidingSyncProxySupported(): Boolean {
return client.availableSlidingSyncVersions().any { it is SlidingSyncVersion.Proxy }
}
override fun isUsingNativeSlidingSync(): Boolean {
return client.session().slidingSyncVersion == SlidingSyncVersion.Native
override suspend fun currentSlidingSyncVersion(): Result<SlidingSyncVersion> = withContext(sessionDispatcher) {
runCatching {
innerClient.session().slidingSyncVersion.map()
}
}
private suspend fun File.getCacheSize(

View file

@ -77,12 +77,12 @@ class RustMatrixClientFactory @Inject constructor(
.finish()
return RustMatrixClient(
client = client,
innerClient = client,
baseDirectory = baseDirectory,
sessionStore = sessionStore,
appCoroutineScope = appCoroutineScope,
sessionDelegate = sessionDelegate,
syncService = syncService,
innerSyncService = syncService,
dispatchers = coroutineDispatchers,
baseCacheDirectory = cacheDirectory,
clock = clock,

View file

@ -94,10 +94,6 @@ internal class RustEncryptionService(
}
.stateIn(sessionCoroutineScope, SharingStarted.Eagerly, false)
fun destroy() {
service.destroy()
}
override suspend fun enableBackups(): Result<Unit> = withContext(dispatchers.io) {
runCatching {
service.enableBackups()

View file

@ -42,7 +42,6 @@ class RustNotificationSettingsService(
fun destroy() {
notificationSettings.setDelegate(null)
notificationSettings.destroy()
}
override suspend fun getRoomNotificationSettings(roomId: RoomId, isEncrypted: Boolean, isOneToOne: Boolean): Result<RoomNotificationSettings> =

View file

@ -10,12 +10,14 @@ package io.element.android.libraries.matrix.impl.sync
import io.element.android.libraries.matrix.api.sync.SyncService
import io.element.android.libraries.matrix.api.sync.SyncState
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.withContext
import org.matrix.rustcomponents.sdk.SyncServiceState
import timber.log.Timber
import java.util.concurrent.atomic.AtomicBoolean
@ -49,12 +51,11 @@ class RustSyncService(
Timber.d("Stop sync failed: $it")
}
suspend fun destroy() {
suspend fun destroy() = withContext(NonCancellable) {
// If the service was still running, stop it
stopSync()
Timber.d("Destroying sync service")
isServiceReady.set(false)
innerSyncService.destroy()
}
override val syncState: StateFlow<SyncState> =

View file

@ -0,0 +1,19 @@
/*
* Copyright 2024 New Vector Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only
* Please see LICENSE in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.sync
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import org.matrix.rustcomponents.sdk.SlidingSyncVersion as RustSlidingSyncVersion
internal fun RustSlidingSyncVersion.map(): SlidingSyncVersion {
return when (this) {
RustSlidingSyncVersion.None -> SlidingSyncVersion.None
is RustSlidingSyncVersion.Proxy -> SlidingSyncVersion.Proxy
RustSlidingSyncVersion.Native -> SlidingSyncVersion.Native
}
}

View file

@ -229,7 +229,6 @@ class RustSessionVerificationService(
recoveryStateListenerTaskHandle.cancelAndDestroy()
if (this::verificationController.isInitialized) {
verificationController.setDelegate(null)
verificationController.destroy()
}
}

View file

@ -35,14 +35,14 @@ class RustMatrixClientTest {
private fun TestScope.createRustMatrixClient(
sessionStore: SessionStore = InMemorySessionStore(),
) = RustMatrixClient(
client = FakeRustClient(),
innerClient = FakeRustClient(),
baseDirectory = File(""),
sessionStore = sessionStore,
appCoroutineScope = this,
sessionDelegate = aRustClientSessionDelegate(
sessionStore = sessionStore,
),
syncService = FakeRustSyncService(),
innerSyncService = FakeRustSyncService(),
dispatchers = testCoroutineDispatchers(),
baseCacheDirectory = File(""),
clock = FakeSystemClock(),

View file

@ -30,6 +30,7 @@ import io.element.android.libraries.matrix.api.room.preview.RoomPreviewInfo
import io.element.android.libraries.matrix.api.roomdirectory.RoomDirectoryService
import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.roomlist.RoomSummary
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.user.MatrixSearchUserResults
import io.element.android.libraries.matrix.api.user.MatrixUser
import io.element.android.libraries.matrix.api.verification.SessionVerificationService
@ -84,9 +85,8 @@ class FakeMatrixClient(
private val getUrlLambda: (String) -> Result<String> = { lambdaError() },
private val canDeactivateAccountResult: () -> Boolean = { lambdaError() },
private val deactivateAccountResult: (String, Boolean) -> Result<Unit> = { _, _ -> lambdaError() },
var isNativeSlidingSyncSupportedLambda: suspend () -> Boolean = { true },
var isSlidingSyncProxySupportedLambda: suspend () -> Boolean = { true },
var isUsingNativeSlidingSyncLambda: () -> Boolean = { true },
private val currentSlidingSyncVersionLambda: () -> Result<SlidingSyncVersion> = { lambdaError() },
private val availableSlidingSyncVersionsLambda: () -> Result<List<SlidingSyncVersion>> = { lambdaError() }
) : MatrixClient {
var setDisplayNameCalled: Boolean = false
private set
@ -340,15 +340,11 @@ class FakeMatrixClient(
return getUrlLambda(url)
}
override suspend fun isNativeSlidingSyncSupported(): Boolean {
return isNativeSlidingSyncSupportedLambda()
override suspend fun currentSlidingSyncVersion(): Result<SlidingSyncVersion> {
return currentSlidingSyncVersionLambda()
}
override suspend fun isSlidingSyncProxySupported(): Boolean {
return isSlidingSyncProxySupportedLambda()
}
override fun isUsingNativeSlidingSync(): Boolean {
return isUsingNativeSlidingSyncLambda()
override suspend fun availableSlidingSyncVersions(): Result<List<SlidingSyncVersion>> {
return availableSlidingSyncVersionsLambda()
}
}