Add MessageBufferingConfig to allow custom back-pressure config for message.new events

[VelikovPetar]

Goal

High-traffic channel types (e.g. livestreams) can produce a flood of message.new events that arrive faster than the sequential event-handling pipeline can process them. The current implementation funnels every socket event through a single MutableSharedFlow with extraBufferCapacity = Int.MAX_VALUE, which means there is no back-pressure: a burst of new-message events queues up unbounded memory and starves more important signals (reads, bans, member updates) of timely processing.

This PR introduces a MessageBufferConfig that lets integrators opt specific channel types into a bounded buffer for NewMessageEvents, with a configurable overflow strategy (SUSPEND / DROP_OLDEST / DROP_LATEST). Signal-critical events and events for non-opted-in channel types are unaffected.

Implementation

• Added MessageBufferConfig (under MessageLimitConfig.messageBufferConfig) exposing:
  • channelTypes: Set<String> — channel types whose NewMessageEvents go through the bounded buffer (empty by default → feature is a no-op).
  • capacity: Int — buffer capacity (defaults to Int.MAX_VALUE).
  • overflow: BufferOverflow — overflow strategy (defaults to SUSPEND).
• EventHandlerSequential now allocates a secondary MutableSharedFlow (bufferedNewMessageEvents) lazily, only when buffering is enabled, so the default configuration pays no cost for it.
• Two listener variants:
  • defaultSocketEventListener — the existing unbuffered path; used when no channel types are opted in.
  • bufferedSocketEventListener — routes NewMessageEvents for opted-in channel types to the bounded flow, and everything else (including non-opted-in NewMessageEvents and all other event types) to the unbuffered flow.
• startListening() picks the listener based on bufferConfig.channelTypes.isNotEmpty() and only collects from bufferedNewMessageEvents when buffering is enabled.
• StreamStatePluginFactory wires the config from StatePluginConfig.messageLimitConfig.messageBufferConfig into EventHandlerSequential.

The bounded flow shares the same downstream pipeline (socketEventCollector → handleBatchEvent) as the unbuffered flow, so ordering inside each flow is preserved and back-pressure is applied independently per flow.

Testing

• Added 200+ lines of unit tests in EventHandlerSequentialTest covering:
  • default behaviour (no buffering) is unchanged when channelTypes is empty.
  • NewMessageEvents for opted-in channel types are routed through the bounded buffer.
  • NewMessageEvents for non-opted-in channel types and all non-NewMessageEvent events keep using the unbuffered path.
  • DROP_OLDEST / DROP_LATEST / SUSPEND overflow strategies behave as expected when the buffer is full.
• Existing tests (TotalUnreadCountTest, EventHandlerSequentialUserMessagesDeletedTest) updated to pass the new bufferConfig argument.

Summary by CodeRabbit

Release Notes

• New Features
  • Introduced customizable message buffering for improved event stream management. Configure buffer settings per channel type with adjustable capacity and overflow handling strategies, enabling optimized message delivery performance and efficient resource management based on specific channel requirements.

[github-actions]

PR checklist ✅

All required conditions are satisfied:

• Title length is OK (or ignored by label).
• At least one pr: label exists.
• Sections ### Goal, ### Implementation, and ### Testing are filled (or ignored for dependabot PRs).

🎉 Great job! This PR is ready for review.

[github-actions]

SDK Size Comparison 📏

SDK	Before	After	Difference	Status
stream-chat-android-client	5.26 MB	5.26 MB	0.00 MB	🟢
stream-chat-android-offline	5.49 MB	5.49 MB	0.00 MB	🟢
stream-chat-android-ui-components	10.64 MB	10.66 MB	0.02 MB	🟢
stream-chat-android-compose	12.87 MB	12.87 MB	0.00 MB	🟢

[github-actions]

This pull request has been automatically marked as stale because it has been inactive for 14 days. It will be closed in 7 days if no further activity occurs.

[coderabbitai]

Walkthrough

This PR extends message limit configuration with per-channel-type buffering control. A new MessageBufferConfig type and updated MessageLimitConfig define overflow policies; EventHandlerSequential implements dual-path event routing; the factory threads the configuration through; and comprehensive overflow tests validate the behavior under DROP_OLDEST, DROP_LATEST, and SUSPEND policies.

Changes

Message Buffering Configuration and Implementation

Layer / File(s)	Summary
Public Configuration Contract stream-chat-android-state/src/main/java/io/getstream/chat/android/state/plugin/config/StatePluginConfig.kt, stream-chat-android-state/api/stream-chat-android-state.api	New MessageBufferConfig data class exposes channelTypes, capacity, and overflow strategy (defaulting to BufferOverflow.SUSPEND). MessageLimitConfig extends to include a messageBufferConfig property. Both contracts surfaced in the public API.
Event Handler Buffering Implementation stream-chat-android-state/src/main/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequential.kt	Constructor accepts bufferConfig parameter. Dual-path event routing: NewMessageEvents for channels in bufferConfig.channelTypes route through a lazily-allocated bounded MutableSharedFlow; others use unbuffered flow. Centralized logEmitOutcome helper. startListening collects from both flows, gated on initJob completion.
Factory Integration stream-chat-android-state/src/main/java/io/getstream/chat/android/state/plugin/factory/StreamStatePluginFactory.kt	Extracts messageBufferConfig from state plugin config and threads it through createEventHandler into EventHandlerSequential constructor.
Test Fixture Wiring stream-chat-android-state/src/test/java/io/getstream/chat/android/state/event/TotalUnreadCountTest.kt, stream-chat-android-state/src/test/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequentialUserMessagesDeletedTest.kt	Both test classes wire bufferConfig = MessageBufferConfig() into their EventHandlerSequential constructions to satisfy updated handler signatures.
Buffer Overflow Test Coverage stream-chat-android-state/src/test/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequentialTest.kt	Parameterized tests validate DROP_OLDEST, DROP_LATEST, and non-buffered channel behavior, including high-volume stress scenarios. Fixture extended with configurable bufferConfig, side-effect control gates, listener capture, and event generation helpers to exercise overflow deterministically.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~22 minutes

Poem

🐰 A bounded flow for messages bright,
With overflow policies set just right!
DROP the oldest, or the new,
Or SUSPEND until there's room for you.
Channels buffered, tests complete—
The rabbit's buffering is so neat! 🎉

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 4.35% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'Add MessageBufferingConfig...' accurately describes the main change: introducing a new configuration class for message event buffering with back-pressure control.
Description check	✅ Passed	The description covers Goal, Implementation, and Testing sections as per template. However, UI Changes, Contributor Checklist, Reviewer Checklist, and GIF sections are missing/incomplete.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feature/AND-1166_add_buffering_support_for_message_new

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (1)

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

211-222: 💤 Low value

Potential division by zero in ratio calculation.

When collectedCount is 0 (before any events are collected), the ratio calculation eCount.toDouble() / cCount.toDouble() will produce Infinity. While this won't crash (Kotlin/JVM handles it gracefully), the logged ratio value will be misleading.

🔧 Suggested fix

 private fun logEmitOutcome(event: ChatEvent, emitted: Boolean) {
     if (emitted) {
         val cCount = collectedCount.get()
         val eCount = emittedCount.incrementAndGet()
-        val ratio = eCount.toDouble() / cCount.toDouble()
+        val ratio = if (cCount > 0) eCount.toDouble() / cCount.toDouble() else 0.0
         StreamLog.v(TAG_SOCKET) {
             "[onSocketEventReceived] event.type: ${event.realType}; $eCount => $cCount ($ratio)"
         }
     } else {
         StreamLog.e(TAG_SOCKET) { "[onSocketEventReceived] failed to emit socket event: $event" }
     }
 }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@stream-chat-android-state/src/main/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequential.kt`
around lines 211 - 222, In logEmitOutcome, avoid dividing by zero when computing
ratio using collectedCount; change the ratio calculation in the emitted branch
of logEmitOutcome (which reads collectedCount.get() and
emittedCount.incrementAndGet()) to compute ratio only when cCount > 0 (e.g.,
ratio = if (cCount > 0) eCount.toDouble() / cCount.toDouble() else 0.0) so the
logged value is meaningful and not Infinity; keep the existing increments and
logging call to StreamLog.v(TAG_SOCKET) but use the guarded ratio variable.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In
`@stream-chat-android-state/src/main/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequential.kt`:
- Around line 211-222: In logEmitOutcome, avoid dividing by zero when computing
ratio using collectedCount; change the ratio calculation in the emitted branch
of logEmitOutcome (which reads collectedCount.get() and
emittedCount.incrementAndGet()) to compute ratio only when cCount > 0 (e.g.,
ratio = if (cCount > 0) eCount.toDouble() / cCount.toDouble() else 0.0) so the
logged value is meaningful and not Infinity; keep the existing increments and
logging call to StreamLog.v(TAG_SOCKET) but use the guarded ratio variable.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: 7572b91e-7416-4e90-a452-8fd4d36ac732

📥 Commits

Reviewing files that changed from the base of the PR and between d404872 and 1967ff3.

📒 Files selected for processing (7)

• stream-chat-android-state/api/stream-chat-android-state.api
• stream-chat-android-state/src/main/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequential.kt
• stream-chat-android-state/src/main/java/io/getstream/chat/android/state/plugin/config/StatePluginConfig.kt
• stream-chat-android-state/src/main/java/io/getstream/chat/android/state/plugin/factory/StreamStatePluginFactory.kt
• stream-chat-android-state/src/test/java/io/getstream/chat/android/state/event/TotalUnreadCountTest.kt
• stream-chat-android-state/src/test/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequentialTest.kt
• stream-chat-android-state/src/test/java/io/getstream/chat/android/state/event/handler/internal/EventHandlerSequentialUserMessagesDeletedTest.kt

[sonarqubecloud]

Quality Gate failed

Failed conditions
78.6% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube Cloud