Scope typing indicators to the channel or the thread the user is typing in

stream-chat-android-client/api/stream-chat-android-client.api

@@ -993,6 +993,7 @@ public abstract interface class io/getstream/chat/android/client/api/state/Threa
 	public abstract fun getNewestInThread ()Lkotlinx/coroutines/flow/StateFlow;
 	public abstract fun getOldestInThread ()Lkotlinx/coroutines/flow/StateFlow;
 	public abstract fun getParentId ()Ljava/lang/String;
+	public fun getTyping ()Lkotlinx/coroutines/flow/StateFlow;
 }
 
 public final class io/getstream/chat/android/client/api/state/querychannels/GroupedQueryConfig {

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/state/ThreadState.kt

@@ -18,7 +18,10 @@ package io.getstream.chat.android.client.api.state
 
 import io.getstream.chat.android.core.internal.InternalStreamChatApi
 import io.getstream.chat.android.models.Message
+import io.getstream.chat.android.models.TypingEvent
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
 
 /** State container with reactive data of a thread.*/
 public interface ThreadState {
@@ -32,6 +35,10 @@ public interface ThreadState {
     /** If we are currently loading messages. */
     public val loading: StateFlow<Boolean>
 
+    /** The users currently typing a reply in this thread, other than the current user. */
+    public val typing: StateFlow<TypingEvent>
+        get() = noTyping
+
     /** If we've reached the earliest point in this thread. */
     public val endOfOlderMessages: StateFlow<Boolean>
 
@@ -55,3 +62,6 @@ public interface ThreadState {
     public val isJumpPage: Boolean
         get() = false
 }
+
+private val noTyping: StateFlow<TypingEvent> =
+    MutableStateFlow(TypingEvent(channelId = "", users = emptyList())).asStateFlow()

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequential.kt

@@ -76,6 +76,8 @@ import io.getstream.chat.android.client.events.ReminderCreatedEvent
 import io.getstream.chat.android.client.events.ReminderDeletedEvent
 import io.getstream.chat.android.client.events.ReminderUpdatedEvent
 import io.getstream.chat.android.client.events.ThreadUpdatedEvent
+import io.getstream.chat.android.client.events.TypingStartEvent
+import io.getstream.chat.android.client.events.TypingStopEvent
 import io.getstream.chat.android.client.events.UserEvent
 import io.getstream.chat.android.client.events.UserMessagesDeletedEvent
 import io.getstream.chat.android.client.events.UserPresenceChangedEvent
@@ -586,6 +588,10 @@ internal class EventHandlerSequential(
             .forEach { (messageId, events) ->
                 logicRegistry.thread(messageId).handleReminderEvents(events)
             }
+        // A typing stop can name another thread than its start, so every active thread sees the typing events
+        sortedEvents.filter { it.isTypingFromOtherUser() }.takeIf { it.isNotEmpty() }?.let { events ->
+            logicRegistry.getActiveThreadsLogic().forEach { thread -> thread.handleTypingEvents(events) }
+        }
         sortedEvents.filterIsInstance<HasPoll>()
             .groupBy { it.messageId }
             .filterKeys { it != null && logicRegistry.isActiveThread(it) }
@@ -607,6 +613,12 @@ internal class EventHandlerSequential(
         logger.v { "[updateThreadState] completed batchId: ${batchEvent.id}" }
     }
 
+    private fun ChatEvent.isTypingFromOtherUser(): Boolean = when (this) {
+        is TypingStartEvent -> user.id != currentUserId
+        is TypingStopEvent -> user.id != currentUserId
+        else -> false
+    }
+
     @Suppress("LongMethod", "LoopWithTooManyJumpStatements")
     private suspend fun updateOfflineStorage(batchEvent: BatchEvent) {
         logger.v { "[updateOfflineStorage] batchId: ${batchEvent.id}, batchEvent.size: ${batchEvent.size} " }

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelEventHandlerImpl.kt

@@ -314,13 +314,14 @@ internal class ChannelEventHandlerImpl(
                 state.removeTruncatedMessages(event.truncatedBefore, systemMessage = null)
                 state.updateChannelData(event)
             }
-            // Typing events
+            // Typing events. Typing in a thread only shows in that thread.
             is TypingStartEvent -> {
-                if (event.user.id != getCurrentUserId()) {
+                if (event.parentId == null && event.user.id != getCurrentUserId()) {
                     typingEventPruner.processEvent(event.user.id, event)
                 }
             }
 
+            // A user types in one place at a time, so a stop clears them whichever thread it names
             is TypingStopEvent -> {
                 if (event.user.id != getCurrentUserId()) {
                     typingEventPruner.processEvent(event.user.id, null)

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelEventHandlerLegacyImpl.kt

@@ -233,8 +233,9 @@ internal class ChannelEventHandlerLegacyImpl(
                 stateLogic.removeTruncatedMessages(event.truncatedBefore)
                 stateLogic.updateChannelData(event)
             }
-            // Typing events
-            is TypingStartEvent -> stateLogic.setTyping(event.user.id, event)
+            // Typing events. Typing in a thread only shows in that thread.
+            is TypingStartEvent -> if (event.parentId == null) stateLogic.setTyping(event.user.id, event)
+            // A user types in one place at a time, so a stop clears them whichever thread it names
             is TypingStopEvent -> stateLogic.setTyping(event.user.id, null)
             // Read/delivery receipt events
             is MessageReadEvent -> {

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/thread/internal/ThreadLogic.kt

@@ -17,6 +17,7 @@
 package io.getstream.chat.android.client.internal.state.plugin.logic.channel.thread.internal
 
 import io.getstream.chat.android.client.events.AnswerCastedEvent
+import io.getstream.chat.android.client.events.ChatEvent
 import io.getstream.chat.android.client.events.HasMessage
 import io.getstream.chat.android.client.events.HasPoll
 import io.getstream.chat.android.client.events.HasReminder
@@ -31,24 +32,33 @@ import io.getstream.chat.android.client.events.PollUpdatedEvent
 import io.getstream.chat.android.client.events.ReminderCreatedEvent
 import io.getstream.chat.android.client.events.ReminderDeletedEvent
 import io.getstream.chat.android.client.events.ReminderUpdatedEvent
+import io.getstream.chat.android.client.events.TypingStartEvent
+import io.getstream.chat.android.client.events.TypingStopEvent
 import io.getstream.chat.android.client.events.VoteCastedEvent
 import io.getstream.chat.android.client.events.VoteChangedEvent
 import io.getstream.chat.android.client.events.VoteRemovedEvent
 import io.getstream.chat.android.client.extensions.internal.processPoll
 import io.getstream.chat.android.client.extensions.internal.toMessageReminderInfo
+import io.getstream.chat.android.client.internal.state.plugin.logic.channel.internal.TypingEventPruner
 import io.getstream.chat.android.client.internal.state.plugin.state.channel.thread.internal.ThreadMutableState
 import io.getstream.chat.android.client.plugin.listeners.ThreadQueryListener
 import io.getstream.chat.android.models.MemberInfo
 import io.getstream.chat.android.models.Message
 import io.getstream.chat.android.models.Poll
+import kotlinx.coroutines.CoroutineScope
 
 /** Logic class for thread state management. Implements [ThreadQueryListener] as listener for LLC requests. */
+@Suppress("TooManyFunctions")
 internal class ThreadLogic(
     private val threadStateLogic: ThreadStateLogic,
+    private val coroutineScope: CoroutineScope,
 ) {
 
     private val mutableState: ThreadMutableState = threadStateLogic.writeThreadState()
 
+    // Created with the first typing start, which carries the channel id the typing state reports.
+    private var typingEventPruner: TypingEventPruner? = null
+
     fun isLoadingMessages(): Boolean = mutableState.loading.value
 
     internal fun setLoading(isLoading: Boolean) {
@@ -198,6 +208,28 @@ internal class ThreadLogic(
         }
     }
 
+    /**
+     * Applies typing starts sent from this thread, and every typing stop. The caller leaves out the current user's own
+     * events.
+     */
+    internal fun handleTypingEvents(events: List<ChatEvent>) {
+        events.forEach { event ->
+            when (event) {
+                is TypingStartEvent -> if (event.parentId == mutableState.parentId) {
+                    typingEventPruner(event.channelId).processEvent(event.user.id, event)
+                }
+                // A user types in one place at a time, so a stop clears them whichever thread it names
+                is TypingStopEvent -> typingEventPruner?.processEvent(event.user.id, null)
+                else -> Unit
+            }
+        }
+    }
+
+    private fun typingEventPruner(channelId: String): TypingEventPruner = typingEventPruner
+        ?: TypingEventPruner(channelId, coroutineScope) { _, typingEvent ->
+            mutableState.setTyping(typingEvent)
+        }.also { typingEventPruner = it }
+
     internal fun handleReminderEvents(events: List<HasReminder>) {
         val messages = events.mapNotNull { event ->
             val message = getMessage(event.reminder.messageId)

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/internal/LogicRegistry.kt

@@ -313,7 +313,7 @@ internal class LogicRegistry internal constructor(
         return threads.getOrPut(messageId) {
             val mutableState = stateRegistry.mutableThread(messageId)
             val stateLogic = ThreadStateLogic(mutableState)
-            ThreadLogic(stateLogic).also { threadLogic ->
+            ThreadLogic(stateLogic, coroutineScope).also { threadLogic ->
                 coroutineScope.launch {
                     val parentMessage = getMessageById(messageId) ?: repos.selectMessage(messageId)
                     parentMessage?.let { threadLogic.upsertMessage(it) }

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/thread/internal/ThreadMutableState.kt

@@ -22,6 +22,7 @@ import io.getstream.chat.android.client.utils.message.isDeleted
 import io.getstream.chat.android.client.utils.message.isLocalOnly
 import io.getstream.chat.android.models.Message
 import io.getstream.chat.android.models.Poll
+import io.getstream.chat.android.models.TypingEvent
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
@@ -42,6 +43,7 @@ internal class ThreadMutableState(
     private var _endOfNewerMessages: MutableStateFlow<Boolean>? = MutableStateFlow(false)
     private var _oldestInThread: MutableStateFlow<Message?>? = MutableStateFlow(null)
     private var _newestInThread: MutableStateFlow<Message?>? = MutableStateFlow(null)
+    private val _typing = MutableStateFlow(TypingEvent(channelId = "", users = emptyList()))
 
     val rawMessage: StateFlow<Map<String, Message>> = _messages!!
     override val messages: StateFlow<List<Message>> = rawMessage
@@ -53,6 +55,7 @@ internal class ThreadMutableState(
     override val endOfNewerMessages: StateFlow<Boolean> = _endOfNewerMessages!!
     override val oldestInThread: StateFlow<Message?> = _oldestInThread!!
     override val newestInThread: StateFlow<Message?> = _newestInThread!!
+    override val typing: StateFlow<TypingEvent> = _typing
 
     @Volatile
     override var isJumpPage: Boolean = false
@@ -74,6 +77,10 @@ internal class ThreadMutableState(
     val parentMessage: Message?
         get() = _messages?.value?.get(parentId)
 
+    fun setTyping(typingEvent: TypingEvent) {
+        _typing.value = typingEvent
+    }
+
     fun setLoading(isLoading: Boolean) {
         _loading?.value = isLoading
     }

stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/channel/controller/WhenHandleEvent.kt

@@ -219,8 +219,8 @@ internal class WhenHandleEvent : SynchronizedCoroutineTest {
         val user1 = randomUser()
         val user2 = randomUser()
 
-        val typingStartEvent1 = randomTypingStartEvent(user = user1, channelId = channelId)
-        val typingStopEvent = randomTypingStopEvent(user = user2, channelId = channelId)
+        val typingStartEvent1 = randomTypingStartEvent(user = user1, channelId = channelId, parentId = null)
+        val typingStopEvent = randomTypingStopEvent(user = user2, channelId = channelId, parentId = null)
 
         channelLogic.run {
             handleEvent(typingStartEvent1)

stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialThreadTypingTest.kt

@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-chat-android/blob/main/LICENSE
+ *
+ * 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.getstream.chat.android.client.internal.state.event.handler.internal
+
+import io.getstream.chat.android.client.ChatEventListener
+import io.getstream.chat.android.client.api.MessageBufferConfig
+import io.getstream.chat.android.client.api.state.StateRegistry
+import io.getstream.chat.android.client.events.ChatEvent
+import io.getstream.chat.android.client.internal.state.plugin.logic.channel.thread.internal.ThreadLogic
+import io.getstream.chat.android.client.internal.state.plugin.logic.internal.LogicRegistry
+import io.getstream.chat.android.client.internal.state.plugin.state.global.internal.MutableGlobalState
+import io.getstream.chat.android.client.persistance.repository.RepositoryFacade
+import io.getstream.chat.android.client.setup.state.ClientState
+import io.getstream.chat.android.client.test.randomTypingStartEvent
+import io.getstream.chat.android.client.test.randomTypingStopEvent
+import io.getstream.chat.android.client.utils.observable.Disposable
+import io.getstream.chat.android.randomString
+import io.getstream.chat.android.randomUser
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.any
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
+import org.mockito.kotlin.verify
+
+/**
+ * Covers how [EventHandlerSequential] routes typing events to the active threads.
+ */
+internal class EventHandlerSequentialThreadTypingTest {
+
+    private val currentUser = randomUser()
+    private val parentId = randomString()
+    private val threadLogic: ThreadLogic = mock()
+    private val logicRegistry: LogicRegistry = mock {
+        on { getActiveThreadsLogic() } doReturn listOf(threadLogic)
+    }
+
+    @Test
+    fun `Typing events from other users reach the active threads`() = runTest {
+        val start = randomTypingStartEvent(user = randomUser(), parentId = parentId)
+        val stop = randomTypingStopEvent(user = randomUser(), parentId = null)
+
+        createEventHandler(scope = this).handleEvents(start, stop)
+
+        verify(threadLogic).handleTypingEvents(listOf(start, stop))
+    }
+
+    @Test
+    fun `The current user's own typing is not routed`() = runTest {
+        val event = randomTypingStartEvent(user = currentUser, parentId = parentId)
+
+        createEventHandler(scope = this).handleEvents(event)
+
+        verify(threadLogic, never()).handleTypingEvents(any())
+    }
+
+    private fun createEventHandler(scope: CoroutineScope): EventHandlerSequential {
+        val subscribeForEvents: (ChatEventListener<ChatEvent>) -> Disposable = { _ ->
+            EventHandlerSequential.EMPTY_DISPOSABLE
+        }
+        val repos: RepositoryFacade = mock {
+            onBlocking { selectChannels(any()) } doReturn emptyList()
+            onBlocking { selectMessages(any()) } doReturn emptyList()
+            onBlocking { selectThreads(any()) } doReturn emptyList()
+        }
+        val stateRegistry: StateRegistry = mock()
+        val clientState: ClientState = mock {
+            on { user } doReturn MutableStateFlow(currentUser)
+        }
+        val syncedEvents: Flow<List<ChatEvent>> = emptyFlow()
+        return EventHandlerSequential(
+            currentUserId = currentUser.id,
+            subscribeForEvents = subscribeForEvents,
+            logicRegistry = logicRegistry,
+            stateRegistry = stateRegistry,
+            clientState = clientState,
+            mutableGlobalState = MutableGlobalState(currentUser.id),
+            repos = repos,
+            sideEffect = {},
+            syncedEvents = syncedEvents,
+            bufferConfig = MessageBufferConfig(),
+            isLocalUnreadCountEnabled = false,
+            scope = scope,
+        )
+    }
+}

stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ThreadQueryListenerStateTest.kt

@@ -142,7 +142,7 @@ internal class ThreadQueryListenerStateTest {
     fun `given a reply held on a jump page, it is kept when the newest page replaces the jump page`() = runTest {
         val parentId = randomString()
         val state = ThreadMutableState(parentId, backgroundScope)
-        val realThreadLogic = ThreadLogic(ThreadStateLogic(state))
+        val realThreadLogic = ThreadLogic(ThreadStateLogic(state), backgroundScope)
         val registry: LogicRegistry = mock { on(it.thread(parentId)) doReturn realThreadLogic }
         val listener = ThreadQueryListenerState(registry)
         val newestPage = listOf(reply(parentId, createdAt = 20))

stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelEventHandlerImplTest.kt

@@ -1015,7 +1015,7 @@ internal class ChannelEventHandlerImplTest {
     fun `When TypingStartEvent from other user is handled, Then typing is set via pruner`() {
         val otherUserId = randomString()
         val user = randomUser(id = otherUserId)
-        val event = randomTypingStartEvent(cid = cid, user = user)
+        val event = randomTypingStartEvent(cid = cid, user = user, parentId = null)
 
         handler.handle(event)
 
@@ -1030,7 +1030,7 @@ internal class ChannelEventHandlerImplTest {
     @Test
     fun `When TypingStartEvent from current user is handled, Then typing is not set`() {
         val user = randomUser(id = currentUserId)
-        val event = randomTypingStartEvent(cid = cid, user = user)
+        val event = randomTypingStartEvent(cid = cid, user = user, parentId = null)
 
         handler.handle(event)
 
@@ -1042,7 +1042,7 @@ internal class ChannelEventHandlerImplTest {
         val otherUserId = randomString()
         val user = randomUser(id = otherUserId)
         // First send a start event to populate the typing state
-        val startEvent = randomTypingStartEvent(cid = cid, user = user)
+        val startEvent = randomTypingStartEvent(cid = cid, user = user, parentId = null)
         handler.handle(startEvent)
 
         // Then send a stop event
@@ -1063,6 +1063,25 @@ internal class ChannelEventHandlerImplTest {
         verify(state, never()).setTyping(any())
     }
 
+    @Test
+    fun `When TypingStartEvent from a thread is handled, Then channel typing is not set`() {
+        val event = randomTypingStartEvent(cid = cid, user = randomUser(), parentId = randomString())
+
+        handler.handle(event)
+
+        verify(state, never()).setTyping(any())
+    }
+
+    @Test
+    fun `When TypingStopEvent names a thread, Then channel typing is still cleared`() {
+        val user = randomUser()
+        handler.handle(randomTypingStartEvent(cid = cid, user = user, parentId = null))
+
+        handler.handle(randomTypingStopEvent(cid = cid, user = user, parentId = randomString()))
+
+        verify(state).setTyping(TypingEvent(cid.cidToTypeAndId().second, emptyList()))
+    }
+
     // endregion
 
     // region Read/delivery receipt events tests