Merge pull request #998 from vector-im/feature/fga/safer_callback_flows

No crash when room is already destroyed...

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

@@ -16,11 +16,13 @@
 
 package io.element.android.libraries.matrix.impl.room
 
+import io.element.android.libraries.core.data.tryOrNull
 import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.catch
 import org.matrix.rustcomponents.sdk.RoomList
 import org.matrix.rustcomponents.sdk.RoomListEntriesListener
 import org.matrix.rustcomponents.sdk.RoomListEntriesUpdate
@@ -41,8 +43,14 @@ fun RoomList.loadingStateFlow(): Flow<RoomListLoadingState> =
             }
         }
         val result = loadingState(listener)
-        send(result.state)
+        try {
+            send(result.state)
+        } catch (exception: Exception) {
+            Timber.d("loadingStateFlow() initialState failed.")
+        }
         result.stateStream
+    }.catch {
+        Timber.d(it, "loadingStateFlow() failed")
     }.buffer(Channel.UNLIMITED)
 
 fun RoomList.entriesFlow(onInitialList: suspend (List<RoomListEntry>) -> Unit): Flow<List<RoomListEntriesUpdate>> =
@@ -53,8 +61,26 @@ fun RoomList.entriesFlow(onInitialList: suspend (List<RoomListEntry>) -> Unit):
             }
         }
         val result = entries(listener)
-        onInitialList(result.entries)
+        try {
+            onInitialList(result.entries)
+        } catch (exception: Exception) {
+            Timber.d("entriesFlow() onInitialList failed.")
+        }
         result.entriesStream
+    }.catch {
+        Timber.d(it, "entriesFlow() failed")
+    }.buffer(Channel.UNLIMITED)
+
+fun RoomListService.stateFlow(): Flow<RoomListServiceState> =
+    mxCallbackFlow {
+        val listener = object : RoomListServiceStateListener {
+            override fun onUpdate(state: RoomListServiceState) {
+                trySendBlocking(state)
+            }
+        }
+        tryOrNull {
+            state(listener)
+        }
     }.buffer(Channel.UNLIMITED)
 
 fun RoomListService.roomOrNull(roomId: String): RoomListItem? {
@@ -65,13 +91,3 @@ fun RoomListService.roomOrNull(roomId: String): RoomListItem? {
         return null
     }
 }
-
-fun RoomListService.stateFlow(): Flow<RoomListServiceState> =
-    mxCallbackFlow {
-        val listener = object : RoomListServiceStateListener {
-            override fun onUpdate(state: RoomListServiceState) {
-                trySendBlocking(state)
-            }
-        }
-        state(listener)
-    }.buffer(Channel.UNLIMITED)

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

@@ -16,6 +16,7 @@
 
 package io.element.android.libraries.matrix.impl.sync
 
+import io.element.android.libraries.core.data.tryOrNull
 import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.trySendBlocking
@@ -32,5 +33,7 @@ fun SyncService.stateFlow(): Flow<SyncServiceState> =
                 trySendBlocking(state)
             }
         }
-        state(listener)
+        tryOrNull {
+            state(listener)
+        }
     }.buffer(Channel.UNLIMITED)

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

@@ -16,6 +16,8 @@
 
 package io.element.android.libraries.matrix.impl.timeline
 
+import io.element.android.libraries.core.data.tryOrNull
+import io.element.android.libraries.matrix.impl.util.cancelAndDestroy
 import io.element.android.libraries.matrix.impl.util.destroyAll
 import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
 import kotlinx.coroutines.channels.Channel
@@ -24,10 +26,10 @@ import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.flow.catch
 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
@@ -35,26 +37,26 @@ import timber.log.Timber
 
 internal fun Room.timelineDiffFlow(onInitialList: suspend (List<TimelineItem>) -> Unit): Flow<TimelineDiff> =
     callbackFlow {
-        val roomId = id()
-        Timber.d("Open timelineDiffFlow for room $roomId")
         val listener = object : TimelineListener {
             override fun onUpdate(diff: TimelineDiff) {
                 trySendBlocking(diff)
             }
         }
-        var result: RoomTimelineListenerResult? = null
+        val roomId = id()
+        Timber.d("Open timelineDiffFlow for room $roomId")
+        val result = addTimelineListener(listener)
         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.cancelAndDestroy()
-            result?.itemsStream?.destroy()
+            result.items.destroyAll()
-            result?.items?.destroyAll()
         }
+    }.catch {
+        Timber.d(it, "timelineDiffFlow() failed")
     }.buffer(Channel.UNLIMITED)
 
 internal fun Room.backPaginationStatusFlow(): Flow<BackPaginationStatus> =
@@ -64,5 +66,7 @@ internal fun Room.backPaginationStatusFlow(): Flow<BackPaginationStatus> =
                 trySendBlocking(status)
             }
         }
-        subscribeToBackPaginationStatus(listener)
+        tryOrNull {
+            subscribeToBackPaginationStatus(listener)
+        }
     }.buffer(Channel.UNLIMITED)

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

@@ -21,11 +21,10 @@ import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.callbackFlow
 import org.matrix.rustcomponents.sdk.TaskHandle
 
-internal fun <T> mxCallbackFlow(block: suspend ProducerScope<T>.() -> TaskHandle) =
+internal fun <T> mxCallbackFlow(block: suspend ProducerScope<T>.() -> TaskHandle?) =
     callbackFlow {
-        val token: TaskHandle = block(this)
+        val taskHandle: TaskHandle? = block(this)
         awaitClose {
-            token.cancel()
+            taskHandle?.cancelAndDestroy()
-            token.destroy()
         }
     }

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

@@ -19,18 +19,22 @@ package io.element.android.libraries.matrix.impl.util
 import org.matrix.rustcomponents.sdk.TaskHandle
 import java.util.concurrent.CopyOnWriteArraySet
 
-class TaskHandleBag(private val tokens: MutableSet<TaskHandle> = CopyOnWriteArraySet()) : Set<TaskHandle> by tokens {
+fun TaskHandle.cancelAndDestroy() {
+    cancel()
+    destroy()
+}
+
+class TaskHandleBag(private val taskHandles: MutableSet<TaskHandle> = CopyOnWriteArraySet()) : Set<TaskHandle> by taskHandles {
 
     operator fun plusAssign(taskHandle: TaskHandle?) {
         if (taskHandle == null) return
-        tokens += taskHandle
+        taskHandles += taskHandle
     }
 
     fun dispose() {
-        tokens.forEach {
+        taskHandles.forEach {
-            it.cancel()
+            it.cancelAndDestroy()
-            it.destroy()
         }
-        tokens.clear()
+        taskHandles.clear()
     }
 }