gpunto · Oct 21, 2026 · Oct 21, 2026 · Oct 21, 2026 · Oct 21, 2026
@@ -29,7 +29,7 @@ import kotlinx.coroutines.flow.StateFlow
  * @param controller The [ThreadListController] handling the business logic and the state management for the threads
  * list.
  */
-public class ThreadListViewModel(private val controller: ThreadListController) : ViewModel() {
+public class ThreadListViewModel(private val controller: ThreadListController) : ViewModel(controller) {
 
     /**
      * The current thread list state.

@@ -19,12 +19,13 @@ package io.getstream.chat.android.compose.viewmodel.threads
 import androidx.lifecycle.ViewModel
 import io.getstream.chat.android.client.ChatClient
 import io.getstream.chat.android.client.api.models.QueryThreadsRequest
-import kotlinx.coroutines.test.TestScope
+import io.getstream.chat.android.client.setup.state.ClientState
+import io.getstream.chat.android.models.InitializationState
+import kotlinx.coroutines.flow.MutableStateFlow
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertThrows
 import org.junit.Test
 import org.junit.jupiter.api.assertInstanceOf
-import org.mockito.kotlin.any
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.mock
 
@@ -55,9 +56,12 @@ internal class ThreadListViewModelFactoryTest {
 
     private class Fixture {
 
+        private val mockClientState: ClientState = mock {
+            on { user } doReturn MutableStateFlow(null)
+            on { initializationState } doReturn MutableStateFlow(InitializationState.NOT_INITIALIZED)
+        }
         private val mockChatClient: ChatClient = mock {
-            on { clientState } doReturn mock()
-            on { inheritScope(any()) } doReturn TestScope()
+            on { clientState } doReturn mockClientState
         }
 
         fun get() = ThreadsViewModelFactory(

@@ -16,6 +16,9 @@
 
 package io.getstream.chat.android.compose.viewmodel.threads
 
+import androidx.lifecycle.ViewModel
+import androidx.lifecycle.ViewModelProvider
+import androidx.lifecycle.ViewModelStore
 import app.cash.turbine.test
 import io.getstream.chat.android.randomBoolean
 import io.getstream.chat.android.randomInt
@@ -71,6 +74,17 @@ internal class ThreadListViewModelTest {
         fixture.verifyLoadNextPage()
     }
 
+    @Test
+    fun `clearing the view model closes the controller`() {
+        val fixture = Fixture()
+        val store = ViewModelStore()
+        ViewModelProvider(store, fixture.factory())[ThreadListViewModel::class.java]
+
+        store.clear()
+
+        fixture.verifyClose()
+    }
+
     private class Fixture {
         private val mockController: ThreadListController = mock()
 
@@ -86,6 +100,15 @@ internal class ThreadListViewModelTest {
             verify(mockController).loadNextPage()
         }
 
+        fun verifyClose() = apply {
+            verify(mockController).close()
+        }
+
+        fun factory() = object : ViewModelProvider.Factory {
+            @Suppress("UNCHECKED_CAST")
+            override fun <T : ViewModel> create(modelClass: Class<T>): T = get() as T
+        }
+
         fun get() = ThreadListViewModel(
             controller = mockController,
         )

@@ -25,6 +25,8 @@ import io.getstream.chat.android.models.querysort.QuerySorter
 import io.getstream.chat.android.ui.common.model.MessageResult
 import io.getstream.chat.android.ui.common.state.mentions.MentionListEvent
 import io.getstream.chat.android.ui.common.state.mentions.MentionListState
+import io.getstream.chat.android.ui.common.utils.extensions.awaitConnectedUser
+import io.getstream.chat.android.ui.common.utils.extensions.userSessionEnds
 import io.getstream.log.taggedLogger
 import io.getstream.result.Error
 import io.getstream.result.Result
@@ -35,9 +37,8 @@ import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.flow.flatMapLatest
-import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.mapLatest
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.update
@@ -75,14 +76,21 @@ public class MentionListController(
 
     init {
         loadRequests.onStart { emit(Unit) } // Triggers the initial load
-            .flatMapLatest { flowOf(searchMentions()) }
-            .onEach { result ->
-                when (result) {
+            // A new load, or the reset when a user session ends, cancels the previous result's handling too.
+            .mapLatest {
+                when (val result = searchMentions()) {
                     is Result.Success -> onSuccessResult(result.value)
                     is Result.Failure -> onFailureResult(result.value)
                 }
             }
             .launchIn(scope)
+        chatClient.clientState.userSessionEnds()
+            .onEach {
+                // Drops the previous user's mentions and loads the next user's once they're connected.
+                _state.value = MentionListState()
+                loadRequests.tryEmit(Unit)
+            }
+            .launchIn(scope)
     }
 
     /**
@@ -125,7 +133,9 @@ public class MentionListController(
     }
 
     private suspend fun searchMentions(): Result<SearchMessagesResult> {
-        val currentUser = requireNotNull(chatClient.getCurrentUser())
+        // A user who is still connecting, e.g. right after a switch, can't search yet.
+        val currentUser = chatClient.clientState.awaitConnectedUser()
+            ?: return Result.Failure(Error.GenericError("No user is connected"))
         val channelFilter = Filters.`in`("members", listOf(currentUser.id))
         val messageFilter = Filters.contains("mentioned_users.id", currentUser.id)
         val currentState = _state.value

@@ -25,13 +25,17 @@ import io.getstream.chat.android.core.internal.coroutines.DispatcherProvider
 import io.getstream.chat.android.models.ConnectionState
 import io.getstream.chat.android.models.User
 import io.getstream.chat.android.ui.common.state.threads.ThreadListState
+import io.getstream.chat.android.ui.common.utils.extensions.userSessionEnds
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.launch
 
@@ -41,22 +45,26 @@ import kotlinx.coroutines.launch
  *
  * @param query The [QueryThreadsRequest] object containing the parameters for querying threads.
  * @param chatClient The [ChatClient] instance for retrieving the Threads related data.
- * @param scope The [CoroutineScope] on which coroutines should be launched.
- * @param queryThreadsAsState Function that returns a [StateFlow] of [QueryThreadsState].
+ * @param scope The [CoroutineScope] the controller's work runs under, in a child scope that [close] cancels; the given
+ * scope itself is never cancelled. The default one outlives user sessions, so the list can follow the next user, and
+ * runs one task at a time.
+ * @param queryThreadsAsState Function that returns a [StateFlow] of [QueryThreadsState], built in the given scope.
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 @InternalStreamChatApi
 public class ThreadListController(
     private val query: QueryThreadsRequest,
     private val chatClient: ChatClient = ChatClient.instance(),
-    private val scope: CoroutineScope = chatClient.inheritScope { DispatcherProvider.IO },
-    queryThreadsAsState: () -> StateFlow<QueryThreadsState?> = {
+    scope: CoroutineScope = CoroutineScope(DispatcherProvider.IO.limitedParallelism(1)),
+    queryThreadsAsState: (CoroutineScope) -> StateFlow<QueryThreadsState?> = { stateScope ->
         chatClient.queryThreadsAsState(
             request = query,
-            coroutineScope = scope,
+            coroutineScope = stateScope,
         )
     },
-) {
+) : AutoCloseable {
+
+    private val ownScope = CoroutineScope(scope.coroutineContext + SupervisorJob(scope.coroutineContext[Job]))
 
     /**
      * Exposes the current thread list state.
@@ -75,27 +83,40 @@ public class ThreadListController(
      */
     public val user: StateFlow<User?> = chatClient.clientState.user
 
-    private val queryThreadsState = queryThreadsAsState()
+    private val queryThreadsState = queryThreadsAsState(ownScope)
 
     init {
-        scope.launch {
+        ownScope.launch {
             queryThreadsState
-                .filterNotNull()
-                .flatMapLatest {
-                    combine(
-                        it.threads,
-                        it.loading,
-                        it.loadingMore,
-                        it.unseenThreadIds,
-                        it.loadingError,
-                    ) { threads, loading, loadingMore, unseenThreadIds, loadingError ->
-                        ThreadListState(threads, loading, loadingMore, unseenThreadIds.size, loadingError)
-                    }
+                // Null while no user is connected: stop following the previous state.
+                .flatMapLatest { state ->
+                    state?.let {
+                        combine(
+                            it.threads,
+                            it.loading,
+                            it.loadingMore,
+                            it.unseenThreadIds,
+                            it.loadingError,
+                        ) { threads, loading, loadingMore, unseenThreadIds, loadingError ->
+                            ThreadListState(threads, loading, loadingMore, unseenThreadIds.size, loadingError)
+                        }
+                    } ?: emptyFlow()
                 }
                 .collectLatest {
                     _state.value = it
                 }
         }
+        ownScope.launch {
+            // Drops the previous user's threads; the query runs again once the next user is connected.
+            chatClient.clientState.userSessionEnds().collect { _state.value = INITIAL_STATE }
+        }
+    }
+
+    /**
+     * Cancels the work started by this controller, without cancelling the scope it was given.
+     */
+    override fun close() {
+        ownScope.cancel()
     }
 
     /**

@@ -18,8 +18,11 @@ package io.getstream.chat.android.ui.common.utils.extensions
 
 import io.getstream.chat.android.client.setup.state.ClientState
 import io.getstream.chat.android.core.internal.InternalStreamChatApi
+import io.getstream.chat.android.models.InitializationState
+import io.getstream.chat.android.models.User
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 
@@ -38,3 +41,12 @@ public fun ClientState.userSessionEnds(): Flow<Unit> = flow {
         previousUserId = userId
     }
 }
+
+/**
+ * Waits until the client is ready for a user and returns them, or `null` if they disconnected in the meantime.
+ */
+@InternalStreamChatApi
+public suspend fun ClientState.awaitConnectedUser(): User? {
+    initializationState.first { it == InitializationState.COMPLETE }
+    return user.value
+}