Merge pull request #960 from vector-im/feature/fga/avoid_deadlocks

Feature/fga/avoid deadlocks

appnav/src/main/kotlin/io/element/android/appnav/room/LoadingRoomNodeView.kt

@@ -16,19 +16,13 @@
 
 package io.element.android.appnav.room
 
-import androidx.compose.foundation.background
 import androidx.compose.foundation.layout.Box
 import androidx.compose.foundation.layout.Column
 import androidx.compose.foundation.layout.ExperimentalLayoutApi
-import androidx.compose.foundation.layout.Row
-import androidx.compose.foundation.layout.Spacer
 import androidx.compose.foundation.layout.WindowInsets
 import androidx.compose.foundation.layout.consumeWindowInsets
 import androidx.compose.foundation.layout.fillMaxSize
 import androidx.compose.foundation.layout.padding
-import androidx.compose.foundation.layout.size
-import androidx.compose.foundation.layout.width
-import androidx.compose.foundation.shape.CircleShape
 import androidx.compose.material3.ExperimentalMaterial3Api
 import androidx.compose.runtime.Composable
 import androidx.compose.ui.Alignment
@@ -38,7 +32,7 @@ import androidx.compose.ui.tooling.preview.Preview
 import androidx.compose.ui.tooling.preview.PreviewParameter
 import androidx.compose.ui.unit.dp
 import io.element.android.features.networkmonitor.api.ui.ConnectivityIndicatorView
-import io.element.android.libraries.designsystem.atomic.atoms.PlaceholderAtom
+import io.element.android.libraries.designsystem.atomic.molecules.IconTitlePlaceholdersRowMolecule
 import io.element.android.libraries.designsystem.components.avatar.AvatarSize
 import io.element.android.libraries.designsystem.components.button.BackButton
 import io.element.android.libraries.designsystem.preview.ElementPreviewDark
@@ -47,7 +41,6 @@ import io.element.android.libraries.designsystem.theme.components.CircularProgre
 import io.element.android.libraries.designsystem.theme.components.Scaffold
 import io.element.android.libraries.designsystem.theme.components.Text
 import io.element.android.libraries.designsystem.theme.components.TopAppBar
-import io.element.android.libraries.designsystem.theme.placeholderBackground
 import io.element.android.libraries.theme.ElementTheme
 import io.element.android.libraries.ui.strings.CommonStrings
 
@@ -102,20 +95,7 @@ private fun LoadingRoomTopBar(
             BackButton(onClick = onBackClicked)
         },
         title = {
-            Row(
-                verticalAlignment = Alignment.CenterVertically
-            ) {
-                Box(
-                    modifier = Modifier
-                        .size(AvatarSize.TimelineRoom.dp)
-                        .align(Alignment.CenterVertically)
-                        .background(color = ElementTheme.colors.placeholderBackground, shape = CircleShape)
-                )
-                Spacer(modifier = Modifier.width(8.dp))
-                PlaceholderAtom(width = 20.dp, height = 7.dp)
-                Spacer(modifier = Modifier.width(7.dp))
-                PlaceholderAtom(width = 45.dp, height = 7.dp)
-            }
+            IconTitlePlaceholdersRowMolecule(iconSize = AvatarSize.TimelineRoom.dp)
         },
         windowInsets = WindowInsets(0.dp),
     )

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/MessagesPresenter.kt

@@ -24,7 +24,6 @@ import androidx.compose.runtime.collectAsState
 import androidx.compose.runtime.derivedStateOf
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.mutableStateOf
-import androidx.compose.runtime.produceState
 import androidx.compose.runtime.remember
 import androidx.compose.runtime.rememberCoroutineScope
 import androidx.compose.runtime.saveable.rememberSaveable
@@ -76,6 +75,7 @@ import io.element.android.libraries.matrix.ui.room.canSendMessageAsState
 import io.element.android.libraries.textcomposer.MessageComposerMode
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 import timber.log.Timber
 
 class MessagesPresenter @AssistedInject constructor(
@@ -109,11 +109,13 @@ class MessagesPresenter @AssistedInject constructor(
 
         val syncUpdateFlow = room.syncUpdateFlow.collectAsState()
         val userHasPermissionToSendMessage by room.canSendMessageAsState(type = MessageEventType.ROOM_MESSAGE, updateKey = syncUpdateFlow.value)
-        val roomName by produceState(initialValue = room.displayName, key1 = syncUpdateFlow.value) {
-            value = room.displayName
-        }
-        val roomAvatar by produceState(initialValue = room.avatarData(), key1 = syncUpdateFlow.value) {
-            value = room.avatarData()
+        var roomName: Async<String> by remember { mutableStateOf(Async.Uninitialized) }
+        var roomAvatar: Async<AvatarData> by remember { mutableStateOf(Async.Uninitialized) }
+        LaunchedEffect(syncUpdateFlow.value) {
+            withContext(dispatchers.io) {
+                roomName = Async.Success(room.displayName)
+                roomAvatar = Async.Success(room.avatarData())
+            }
         }
         var hasDismissedInviteDialog by rememberSaveable {
             mutableStateOf(false)

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/MessagesState.kt

@@ -30,8 +30,8 @@ import io.element.android.libraries.matrix.api.core.RoomId
 @Immutable
 data class MessagesState(
     val roomId: RoomId,
-    val roomName: String,
-    val roomAvatar: AvatarData,
+    val roomName: Async<String>,
+    val roomAvatar: Async<AvatarData>,
     val userHasPermissionToSendMessage: Boolean,
     val composerState: MessageComposerState,
     val timelineState: TimelineState,

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/MessagesStateProvider.kt

@@ -38,13 +38,17 @@ open class MessagesStateProvider : PreviewParameterProvider<MessagesState> {
             aMessagesState().copy(composerState = aMessageComposerState().copy(showAttachmentSourcePicker = true)),
             aMessagesState().copy(userHasPermissionToSendMessage = false),
             aMessagesState().copy(showReinvitePrompt = true),
+            aMessagesState().copy(
+                roomName = Async.Uninitialized,
+                roomAvatar = Async.Uninitialized,
+            ),
         )
 }
 
 fun aMessagesState() = MessagesState(
     roomId = RoomId("!id:domain"),
-    roomName = "Room name",
-    roomAvatar = AvatarData("!id:domain", "Room name", size = AvatarSize.TimelineRoom),
+    roomName = Async.Success("Room name"),
+    roomAvatar = Async.Success(AvatarData("!id:domain", "Room name", size = AvatarSize.TimelineRoom)),
     userHasPermissionToSendMessage = true,
     composerState = aMessageComposerState().copy(
         text = "Hello",

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/MessagesView.kt

@@ -62,10 +62,12 @@ import io.element.android.features.messages.impl.timeline.components.retrysendme
 import io.element.android.features.messages.impl.timeline.model.TimelineItem
 import io.element.android.features.networkmonitor.api.ui.ConnectivityIndicatorView
 import io.element.android.libraries.androidutils.ui.hideKeyboard
+import io.element.android.libraries.designsystem.atomic.molecules.IconTitlePlaceholdersRowMolecule
 import io.element.android.libraries.designsystem.components.ProgressDialog
 import io.element.android.libraries.designsystem.components.ProgressDialogType
 import io.element.android.libraries.designsystem.components.avatar.Avatar
 import io.element.android.libraries.designsystem.components.avatar.AvatarData
+import io.element.android.libraries.designsystem.components.avatar.AvatarSize
 import io.element.android.libraries.designsystem.components.button.BackButton
 import io.element.android.libraries.designsystem.components.dialogs.ConfirmationDialog
 import io.element.android.libraries.designsystem.preview.ElementPreviewDark
@@ -135,8 +137,8 @@ fun MessagesView(
             Column {
                 ConnectivityIndicatorView(isOnline = state.hasNetworkConnection)
                 MessagesViewTopBar(
-                    roomTitle = state.roomName,
-                    roomAvatar = state.roomAvatar,
+                    roomName = state.roomName.dataOrNull(),
+                    roomAvatar = state.roomAvatar.dataOrNull(),
                     onBackPressed = onBackPressed,
                     onRoomDetailsClicked = onRoomDetailsClicked,
                 )
@@ -201,7 +203,7 @@ fun MessagesView(
 }
 
 @Composable
-fun ReinviteDialog(state: MessagesState) {
+private fun ReinviteDialog(state: MessagesState) {
     if (state.showReinvitePrompt) {
         ConfirmationDialog(
             title = stringResource(id = R.string.screen_room_invite_again_alert_title),
@@ -238,7 +240,7 @@ private fun AttachmentStateView(
 }
 
 @Composable
-fun MessagesViewContent(
+private fun MessagesViewContent(
     state: MessagesState,
     onMessageClicked: (TimelineItem.Event) -> Unit,
     onUserDataClicked: (UserId) -> Unit,
@@ -286,9 +288,9 @@ fun MessagesViewContent(
 
 @OptIn(ExperimentalMaterial3Api::class)
 @Composable
-fun MessagesViewTopBar(
-    roomTitle: String,
-    roomAvatar: AvatarData,
+private fun MessagesViewTopBar(
+    roomName: String?,
+    roomAvatar: AvatarData?,
     modifier: Modifier = Modifier,
     onRoomDetailsClicked: () -> Unit = {},
     onBackPressed: () -> Unit = {},
@@ -299,17 +301,17 @@ fun MessagesViewTopBar(
             BackButton(onClick = onBackPressed)
         },
         title = {
-            Row(
-                modifier = Modifier.clickable { onRoomDetailsClicked() },
-                verticalAlignment = Alignment.CenterVertically
-            ) {
-                Avatar(roomAvatar)
-                Spacer(modifier = Modifier.width(8.dp))
-                Text(
-                    text = roomTitle,
-                    style = ElementTheme.typography.fontBodyLgMedium,
-                    maxLines = 1,
-                    overflow = TextOverflow.Ellipsis
+            val titleModifier = Modifier.clickable { onRoomDetailsClicked() }
+            if (roomName != null && roomAvatar != null) {
+                RoomAvatarAndNameRow(
+                    roomName = roomName,
+                    roomAvatar = roomAvatar,
+                    modifier = titleModifier
+                )
+            } else {
+                IconTitlePlaceholdersRowMolecule(
+                    iconSize = AvatarSize.TimelineRoom.dp,
+                    modifier = titleModifier
                 )
             }
         },
@@ -318,7 +320,28 @@ fun MessagesViewTopBar(
 }
 
 @Composable
-fun CantSendMessageBanner(
+private fun RoomAvatarAndNameRow(
+    roomName: String,
+    roomAvatar: AvatarData,
+    modifier: Modifier = Modifier
+) {
+    Row(
+        modifier = modifier,
+        verticalAlignment = Alignment.CenterVertically
+    ) {
+        Avatar(roomAvatar)
+        Spacer(modifier = Modifier.width(8.dp))
+        Text(
+            text = roomName,
+            style = ElementTheme.typography.fontBodyLgMedium,
+            maxLines = 1,
+            overflow = TextOverflow.Ellipsis
+        )
+    }
+}
+
+@Composable
+private fun CantSendMessageBanner(
     modifier: Modifier = Modifier,
 ) {
     Row(

features/messages/impl/src/test/kotlin/io/element/android/features/messages/MessagesPresenterTest.kt

@@ -65,6 +65,8 @@ import io.element.android.libraries.mediapickers.test.FakePickerProvider
 import io.element.android.libraries.mediaupload.api.MediaSender
 import io.element.android.libraries.mediaupload.test.FakeMediaPreProcessor
 import io.element.android.libraries.textcomposer.MessageComposerMode
+import io.element.android.tests.testutils.consumeItemsUntilPredicate
+import io.element.android.tests.testutils.consumeItemsUntilTimeout
 import io.element.android.tests.testutils.testCoroutineDispatchers
 import io.mockk.mockk
 import kotlinx.coroutines.test.TestScope
@@ -132,7 +134,6 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
             val initialState = awaitItem()
             initialState.eventSink.invoke(MessagesEvents.HandleAction(TimelineItemAction.Forward, aMessageEvent()))
             assertThat(awaitItem().actionListState.target).isEqualTo(ActionListState.Target.None)
@@ -177,7 +178,6 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
             val initialState = awaitItem()
             initialState.eventSink.invoke(MessagesEvents.HandleAction(TimelineItemAction.Reply, aMessageEvent(eventId = null)))
             assertThat(awaitItem().actionListState.target).isEqualTo(ActionListState.Target.None)
@@ -314,7 +314,6 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
             val initialState = awaitItem()
             initialState.eventSink.invoke(MessagesEvents.HandleAction(TimelineItemAction.ReportContent, aMessageEvent()))
             assertThat(awaitItem().actionListState.target).isEqualTo(ActionListState.Target.None)
@@ -328,10 +327,10 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
             val initialState = awaitItem()
             initialState.eventSink.invoke(MessagesEvents.Dismiss)
             assertThat(awaitItem().actionListState.target).isEqualTo(ActionListState.Target.None)
+
         }
     }
 
@@ -342,7 +341,6 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
             val initialState = awaitItem()
             initialState.eventSink.invoke(MessagesEvents.HandleAction(TimelineItemAction.Developer, aMessageEvent()))
             assertThat(awaitItem().actionListState.target).isEqualTo(ActionListState.Target.None)
@@ -419,9 +417,7 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
-            val initialState = awaitItem()
-            skipItems(1)
+            val initialState = consumeItemsUntilTimeout().last()
             initialState.eventSink(MessagesEvents.InviteDialogDismissed(InviteDialogAction.Invite))
             skipItems(1)
             val loadingState = awaitItem()
@@ -448,9 +444,7 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
-            val initialState = awaitItem()
-            skipItems(1)
+            val initialState = consumeItemsUntilTimeout().last()
             initialState.eventSink(MessagesEvents.InviteDialogDismissed(InviteDialogAction.Invite))
             skipItems(1)
             val loadingState = awaitItem()
@@ -469,9 +463,7 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
-            val initialState = awaitItem()
-            skipItems(1)
+            val initialState = consumeItemsUntilTimeout().last()
             initialState.eventSink(MessagesEvents.InviteDialogDismissed(InviteDialogAction.Invite))
             skipItems(1)
             val loadingState = awaitItem()
@@ -497,15 +489,16 @@ class MessagesPresenterTest {
         moleculeFlow(RecompositionMode.Immediate) {
             presenter.present()
         }.test {
-            skipItems(1)
-            val initialState = awaitItem()
-            skipItems(1)
+            val initialState = consumeItemsUntilTimeout().last()
             initialState.eventSink(MessagesEvents.InviteDialogDismissed(InviteDialogAction.Invite))
-            skipItems(1)
-            val loadingState = awaitItem()
+            val loadingState = consumeItemsUntilPredicate { state ->
+                state.inviteProgress.isLoading()
+            }.last()
             assertThat(loadingState.inviteProgress.isLoading()).isTrue()
-            val newState = awaitItem()
-            assertThat(newState.inviteProgress.isFailure()).isTrue()
+            val failureState = consumeItemsUntilPredicate { state ->
+                state.inviteProgress.isFailure()
+            }.last()
+            assertThat(failureState.inviteProgress.isFailure()).isTrue()
         }
     }
 
@@ -532,8 +525,9 @@ class MessagesPresenterTest {
         }.test {
             // Default value
             assertThat(awaitItem().userHasPermissionToSendMessage).isTrue()
-            skipItems(2)
+            skipItems(1)
             assertThat(awaitItem().userHasPermissionToSendMessage).isFalse()
+            cancelAndIgnoreRemainingEvents()
         }
     }
 

libraries/designsystem/src/main/kotlin/io/element/android/libraries/designsystem/atomic/molecules/IconTitlePlaceholdersRowMolecule.kt

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2023 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.designsystem.atomic.molecules
+
+import androidx.compose.foundation.background
+import androidx.compose.foundation.layout.Arrangement
+import androidx.compose.foundation.layout.Box
+import androidx.compose.foundation.layout.Row
+import androidx.compose.foundation.layout.Spacer
+import androidx.compose.foundation.layout.size
+import androidx.compose.foundation.layout.width
+import androidx.compose.foundation.shape.CircleShape
+import androidx.compose.runtime.Composable
+import androidx.compose.ui.Alignment
+import androidx.compose.ui.Modifier
+import androidx.compose.ui.unit.Dp
+import androidx.compose.ui.unit.dp
+import io.element.android.libraries.designsystem.atomic.atoms.PlaceholderAtom
+import io.element.android.libraries.designsystem.components.avatar.AvatarSize
+import io.element.android.libraries.designsystem.preview.DayNightPreviews
+import io.element.android.libraries.designsystem.preview.ElementPreview
+import io.element.android.libraries.designsystem.theme.placeholderBackground
+import io.element.android.libraries.theme.ElementTheme
+
+@Composable
+fun IconTitlePlaceholdersRowMolecule(
+    iconSize: Dp,
+    modifier: Modifier = Modifier,
+    horizontalArrangement: Arrangement.Horizontal = Arrangement.Start,
+    verticalAlignment: Alignment.Vertical = Alignment.CenterVertically,
+) {
+    Row(
+        modifier = modifier,
+        horizontalArrangement = horizontalArrangement,
+        verticalAlignment = verticalAlignment,
+    ) {
+        Box(
+            modifier = Modifier
+                .size(iconSize)
+                .align(Alignment.CenterVertically)
+                .background(color = ElementTheme.colors.placeholderBackground, shape = CircleShape)
+        )
+        Spacer(modifier = Modifier.width(8.dp))
+        PlaceholderAtom(width = 20.dp, height = 7.dp)
+        Spacer(modifier = Modifier.width(7.dp))
+        PlaceholderAtom(width = 45.dp, height = 7.dp)
+    }
+}
+
+@DayNightPreviews
+@Composable
+internal fun IconTitlePlaceholdersRowMoleculePreview() = ElementPreview {
+    IconTitlePlaceholdersRowMolecule(
+        iconSize = AvatarSize.TimelineRoom.dp,
+    )
+}

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

@@ -150,7 +150,7 @@ class RustMatrixClient constructor(
             }.launchIn(sessionCoroutineScope)
     }
 
-    override suspend fun getRoom(roomId: RoomId): MatrixRoom? {
+    override suspend fun getRoom(roomId: RoomId): MatrixRoom? = withContext(sessionDispatcher) {
         // Check if already in memory...
         var cachedPairOfRoom = pairOfRoom(roomId)
         if (cachedPairOfRoom == null) {
@@ -158,24 +158,24 @@ class RustMatrixClient constructor(
             roomSummaryDataSource.awaitAllRoomsAreLoaded()
             cachedPairOfRoom = pairOfRoom(roomId)
         }
-        if (cachedPairOfRoom == null) return null
-        val (roomListItem, fullRoom) = cachedPairOfRoom
-        return RustMatrixRoom(
-            sessionId = sessionId,
-            roomListItem = roomListItem,
-            innerRoom = fullRoom,
-            sessionCoroutineScope = sessionCoroutineScope,
-            coroutineDispatchers = dispatchers,
-            systemClock = clock,
-            roomContentForwarder = roomContentForwarder,
-            sessionData = sessionStore.getSession(sessionId.value)!!,
-        )
+        cachedPairOfRoom?.let { (roomListItem, fullRoom) ->
+            RustMatrixRoom(
+                sessionId = sessionId,
+                roomListItem = roomListItem,
+                innerRoom = fullRoom,
+                sessionCoroutineScope = sessionCoroutineScope,
+                coroutineDispatchers = dispatchers,
+                systemClock = clock,
+                roomContentForwarder = roomContentForwarder,
+                sessionData = sessionStore.getSession(sessionId.value)!!,
+            )
+        }
     }
 
-    private suspend fun pairOfRoom(roomId: RoomId): Pair<RoomListItem, Room>? = withContext(sessionDispatcher) {
+    private fun pairOfRoom(roomId: RoomId): Pair<RoomListItem, Room>? {
         val cachedRoomListItem = roomListService.roomOrNull(roomId.value)
         val fullRoom = cachedRoomListItem?.fullRoom()
-        if (cachedRoomListItem == null || fullRoom == null) {
+        return if (cachedRoomListItem == null || fullRoom == null) {
             Timber.d("No room cached for $roomId")
             null
         } else {

libraries/matrix/impl/src/main/kotlin/io/element/android/libraries/matrix/impl/timeline/RoomTimelineExtensions.kt

@@ -16,28 +16,45 @@
 
 package io.element.android.libraries.matrix.impl.timeline
 
+import io.element.android.libraries.matrix.impl.util.destroyAll
 import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.callbackFlow
 import org.matrix.rustcomponents.sdk.BackPaginationStatus
 import org.matrix.rustcomponents.sdk.BackPaginationStatusListener
 import org.matrix.rustcomponents.sdk.Room
+import org.matrix.rustcomponents.sdk.RoomTimelineListenerResult
 import org.matrix.rustcomponents.sdk.TimelineDiff
 import org.matrix.rustcomponents.sdk.TimelineItem
 import org.matrix.rustcomponents.sdk.TimelineListener
+import timber.log.Timber
 
 internal fun Room.timelineDiffFlow(onInitialList: suspend (List<TimelineItem>) -> Unit): Flow<TimelineDiff> =
-    mxCallbackFlow {
+    callbackFlow {
+        val roomId = id()
+        Timber.d("Open timelineDiffFlow for room $roomId")
         val listener = object : TimelineListener {
             override fun onUpdate(diff: TimelineDiff) {
                 trySendBlocking(diff)
             }
         }
-        val result = addTimelineListener(listener)
-        onInitialList(result.items)
-        result.itemsStream
+        var result: RoomTimelineListenerResult? = null
+        try {
+            result = addTimelineListener(listener)
+            onInitialList(result.items)
+        } catch (exception: Exception) {
+            Timber.d(exception, "Catch failure in timelineDiffFlow of room $roomId")
+        }
+        awaitClose {
+            Timber.d("Close timelineDiffFlow for room $roomId")
+            result?.itemsStream?.cancel()
+            result?.itemsStream?.destroy()
+            result?.items?.destroyAll()
+        }
     }.buffer(Channel.UNLIMITED)
 
 internal fun Room.backPaginationStatusFlow(): Flow<BackPaginationStatus> =

libraries/matrix/impl/src/main/kotlin/io/element/android/libraries/matrix/impl/timeline/RustMatrixTimeline.kt

@@ -32,6 +32,8 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -131,9 +133,10 @@ class RustMatrixTimeline(
             encryptedHistoryPostProcessor.process(items)
         }
 
-    private suspend fun postItems(items: List<TimelineItem>) {
+    private suspend fun postItems(items: List<TimelineItem>) = coroutineScope {
         // Split the initial items in multiple list as there is no pagination in the cached data, so we can post timelineItems asap.
         items.chunked(INITIAL_MAX_SIZE).reversed().forEach {
+            ensureActive()
             timelineDiffProcessor.postItems(it)
         }
         isInit.set(true)

libraries/matrix/impl/src/main/kotlin/io/element/android/libraries/matrix/impl/util/Disposables.kt

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2023 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.util
+
+import org.matrix.rustcomponents.sdk.Disposable
+
+/**
+ * Call destroy on all elements of the iterable.
+ */
+internal fun Iterable<Disposable>.destroyAll() = forEach { it.destroy() }

tests/testutils/build.gradle.kts

@@ -30,4 +30,5 @@ dependencies {
     implementation(libs.test.junit)
     implementation(libs.coroutines.test)
     implementation(projects.libraries.core)
+    implementation(libs.test.turbine)
 }

tests/testutils/src/main/kotlin/io/element/android/tests/testutils/ReceiveTurbine.kt

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2023 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.tests.testutils
+
+import app.cash.turbine.Event
+import app.cash.turbine.ReceiveTurbine
+import app.cash.turbine.withTurbineTimeout
+import io.element.android.libraries.core.data.tryOrNull
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * Consume all items until timeout is reached waiting for an event or we receive terminal event.
+ * The timeout is applied for each event.
+ * @return the list of consumed items.
+ */
+suspend fun <T : Any> ReceiveTurbine<T>.consumeItemsUntilTimeout(timeout: Duration = 100.milliseconds): List<T> {
+    return consumeItemsUntilPredicate(timeout) { false }
+}
+
+/**
+ * Consume items until predicate is true, or timeout is reached waiting for an event, or we receive terminal event.
+ * The timeout is applied for each event.
+ * @return the list of consumed items.
+ */
+suspend fun <T : Any> ReceiveTurbine<T>.consumeItemsUntilPredicate(
+    timeout: Duration = 100.milliseconds,
+    predicate: (T) -> Boolean,
+): List<T> {
+    val items = ArrayList<T>()
+    tryOrNull {
+        while (true) {
+            when (val event = withTurbineTimeout(timeout) { awaitEvent() }) {
+                is Event.Item<T> -> {
+                    items.add(event.value)
+                    if (predicate(event.value)) {
+                        break
+                    }
+                }
+                else -> break
+            }
+        }
+    }
+    return items
+}