GitHub - GetStream/stream-core-android: Core SDK for the Android products. · GitHub

develop

Go to file

Code

NameName

Last commit message

Last commit date

Repository files navigation

More items

Stream Android Core

Internal Stream SDK This repository is for Stream products only. It is not intended for public consumption or direct integration by third-party apps.

What is Stream?

Stream provides APIs and SDKs for building real-time communication features at scale:

Stream Chat - Production-ready messaging and chat with channels, threads, reactions, moderation, and more
Stream Video - Video calling, audio rooms, and livestreaming with ultra-low latency
Stream Feeds - Scalable activity feeds and social features for apps

Each of these products offers platform-specific SDKs (Android, iOS, React, Flutter, etc.) that developers integrate into their applications.

What is Stream Android Core?

Stream Android Core is the internal foundational library that powers all of Stream's Android SDKs. It's the shared infrastructure layer that provides common functionality needed across Chat, Video, and Feeds:

Purpose

Rather than duplicating infrastructure code across multiple SDKs, Stream Android Core centralizes:

Real-time connectivity: WebSocket management with automatic reconnection and health monitoring
Authentication: Token lifecycle management with automatic refresh
State management: Connection state, network availability, app lifecycle tracking
Reliability: Retry policies, exponential backoff, and connection recovery
Performance: Request deduplication, batching, throttling, serial processing queues
Thread safety: Cross-thread execution utilities and subscription management
Observability: Structured logging and event propagation

Architecture Position

┌─────────────────────────────────────────────────────────┐ │ Application Code (Chat UI, Video Calls, Feeds UI) │ └────────────────────────┬────────────────────────────────┘ │ ┌────────────────────────┴────────────────────────────────┐ │ Stream Product SDKs (Chat SDK, Video SDK, Feeds SDK) │ ← Public APIs └────────────────────────┬────────────────────────────────┘ │ ┌────────────────────────┴────────────────────────────────┐ │ Stream Android Core (This Repository) │ ← Internal Infrastructure │ • WebSocket connections • Serial processing │ │ • Token management • Retry logic │ │ • Lifecycle monitoring • Event batching │ │ • Network detection • Throttling & debouncing │ └────────────────────────┬────────────────────────────────┘ │ ┌────────────────────────┴────────────────────────────────┐ │ Stream Backend APIs (Chat, Video, Feeds) │ └─────────────────────────────────────────────────────────┘

Overview

Stream Android Core provides shared primitives and infrastructure for Stream's Android SDKs:

Authentication & token management
Connection lifecycle & event handling
Network & app lifecycle monitoring
Retry policies with backoff strategies
Serial processing queues
Single-flight execution (request deduplication)
Batching, debouncing & throttling
Thread-safe subscription management
WebSocket connections with health monitoring
Structured logging

API Visibility: Internal vs Published

Stream Android Core uses annotations to distinguish between stable public APIs and internal implementation details.

@StreamPublishedApi - Stable APIs that product SDKs can expose to integrators. These have stability guarantees and follow semantic versioning. Examples: StreamConnectionState, StreamConnectedUser.

@StreamInternalApi - Internal infrastructure not meant for integrators. May change without notice and require @OptIn(StreamInternalApi::class). Examples: StreamRetryProcessor, StreamSerialProcessingQueue, threading utilities.

@StreamDelicateApi - Advanced low-level APIs requiring careful use. Require @OptIn(StreamDelicateApi::class) and generate compiler warnings.

Who Should Use This?

Integrator/app developers: Use Stream Chat SDK, Stream Video SDK, or Stream Feeds SDK. Access published Core types through your product SDK's public API.

Stream SDK maintainers: Use Core as your infrastructure foundation. Expose @StreamPublishedApi types in your SDK's public API, keep @StreamInternalApi types private.

Quick Start

Minimal Setup

Here's a minimal example to get started with Stream Android Core:

import io.getstream.android.core.api.log.StreamLoggerProvider
import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob

// 1. Set up logging
val logProvider = StreamLoggerProvider.defaultAndroidLogger(
    minLevel = StreamLogger.LogLevel.Verbose
)

// 2. Create a coroutine scope
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)

// 3. Create a subscription manager
val subscriptionManager = StreamSubscriptionManager<MyListener>(
    logger = logProvider.taggedLogger("MySubscriptions")
)

Basic Client Instantiation

val singleFlight = StreamSingleFlightProcessor(scope)
val tokenManager = StreamTokenManager(userId, tokenProvider, singleFlight)

val serialQueue = StreamSerialProcessingQueue(
    logger = logProvider.taggedLogger("SerialQueue"),
    scope = scope
)

val client = StreamClient(
    scope = scope,
    apiKey = apiKey,
    userId = userId,
    tokenProvider = tokenProvider,
    logProvider = logProvider,
    tokenManager = tokenManager,
    singleFlight = singleFlight,
    serialQueue = serialQueue,
    // ... other dependencies
)

Requirements

minSdk: 21+ (Android 5.0 Lollipop)
compileSdk/targetSdk: 36
Kotlin: 2.2.0+
Coroutines: 1.10+
AGP: 8.11+
Lifecycle: 2.9+

Core Concepts

Component Lifecycle

Most components in Stream Android Core implement StreamStartableComponent:

interface StreamStartableComponent {
    fun start(): Result<Unit>
    fun stop(): Result<Unit>
}

Pattern: Always call start() before using a component and stop() when done:

val component = StreamSerialProcessingQueue(logger, scope)

component.start()
    .onSuccess { /* component is ready */ }
    .onFailure { error -> logger.e(error) { "Failed to start" } }

// Use the component...

component.stop()
    .onSuccess { /* cleanup complete */ }
    .onFailure { error -> logger.w(error) { "Failed to stop gracefully" } }

Result-Based Error Handling

The library uses Kotlin's Result<T> type consistently:

// Success case
val result: Result<String> = runCatching { "success" }
result.getOrNull() // "success"

// Failure case
val result: Result<String> = runCatching { throw Exception("error") }
result.exceptionOrNull() // Exception("error")

// Handling results
result
    .onSuccess { value -> println(value) }
    .onFailure { error -> logger.e(error) { "Operation failed" } }

Thread Safety

All public APIs are thread-safe unless explicitly documented otherwise
Components use appropriate synchronization internally
Threading utilities ensure operations run on the correct thread (e.g., main thread for lifecycle operations)

Feature Guides

Lifecycle & Network Monitoring

Monitor app lifecycle (foreground/background) and network connectivity changes.

Lifecycle Monitoring

import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor
import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleListener

// Create the monitor
val lifecycleMonitor = StreamLifecycleMonitor(
    logger = logger,
    lifecycle = processLifecycleOwner.lifecycle,
    subscriptionManager = StreamSubscriptionManager(logger)
)

// Subscribe to lifecycle events
val listener = object : StreamLifecycleListener {
    override fun onForeground() {
        logger.i { "App moved to foreground" }
        // Reconnect sockets, resume sync, etc.
    }

    override fun onBackground() {
        logger.i { "App moved to background" }
        // Pause non-critical operations
    }
}

lifecycleMonitor.subscribe(listener).getOrThrow()
lifecycleMonitor.start().getOrThrow()

// Get current state
val state = lifecycleMonitor.getCurrentState()
when (state) {
    is StreamLifecycleState.Foreground -> { /* app is active */ }
    is StreamLifecycleState.Background -> { /* app is backgrounded */ }
    is StreamLifecycleState.Unknown -> { /* not yet determined */ }
}

Network Monitoring

import io.getstream.android.core.api.observers.network.StreamNetworkMonitor
import io.getstream.android.core.api.observers.network.StreamNetworkMonitorListener

val networkMonitor = StreamNetworkMonitor(
    context = context,
    logger = logger,
    subscriptionManager = StreamSubscriptionManager(logger)
)

val listener = object : StreamNetworkMonitorListener {
    override suspend fun onNetworkConnected(snapshot: StreamNetworkInfo.Snapshot?) {
        logger.i { "Network connected: $snapshot" }
        // Retry failed requests, reconnect, etc.
    }

    override suspend fun onNetworkLost(permanent: Boolean) {
        logger.w { "Network lost (permanent: $permanent)" }
        // Queue operations for later, show offline UI
    }
}

networkMonitor.subscribe(listener).getOrThrow()
networkMonitor.start().getOrThrow()

Key Points:

Lifecycle operations (add/remove observers) must run on the main thread
The library handles thread dispatching automatically using runOnMainLooper
Network listener methods are suspend functions and run on background threads

Subscription Management

Thread-safe listener registration with automatic cleanup.

Basic Usage

import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
import io.getstream.android.core.api.subscribe.StreamSubscriptionManager.Options
import io.getstream.android.core.api.subscribe.StreamSubscriptionManager.Options.Retention

// Create manager
val subscriptionManager = StreamSubscriptionManager<MyListener>(
    logger = logger,
    maxStrongSubscriptions = 250, // Strong references
    maxWeakSubscriptions = 250    // Weak references (auto-cleanup)
)

// Subscribe with options
val subscription = subscriptionManager.subscribe(
    listener = myListener,
    options = Options(retention = Retention.KEEP_UNTIL_CANCELLED)
).getOrThrow()

// Notify all listeners
subscriptionManager.forEach { listener ->
    listener.onEvent(event)
}

// Unsubscribe
subscription.cancel()

// Clear all
subscriptionManager.clear()

Retention Policies

enum class Retention {
    // Weak reference - auto-removed when GC'd
    AUTO_REMOVE,

    // Strong reference - kept until explicitly cancelled
    KEEP_UNTIL_CANCELLED
}

When to use each:

AUTO_REMOVE: UI components, fragments, activities (automatic cleanup)
KEEP_UNTIL_CANCELLED: Long-lived services, singletons (explicit lifecycle)

Handling Aggregated Events

When the Event Aggregator is active, listeners registered via StreamSubscriptionManager may receive StreamAggregatedEvent instances during traffic spikes in addition to individual events. Your listener must handle both:

subscriptionManager.forEach { listener ->
    when (event) {
        is StreamAggregatedEvent<*> -> {
            // Batch of events in arrival order — process sequentially
            event.events.forEach { listener.onEvent(it) }
        }
        else -> listener.onEvent(event)
    }
}

See the Event Aggregator section for full details on ordering guarantees and atomic state updates.

Serial Processing Queue

Ordered, single-threaded coroutine pipeline for sequential work.

Basic Usage

import io.getstream.android.core.api.processing.StreamSerialProcessingQueue

val queue = StreamSerialProcessingQueue(
    logger = logger,
    scope = scope
)

queue.start().getOrThrow()

// Submit work - guaranteed to run in order
queue.submit {
    // Work 1 - runs first
    processItem1()
}

queue.submit {
    // Work 2 - runs after work 1 completes
    processItem2()
}

// Stop with timeout
queue.stop(timeout = 5000).getOrThrow()

Use Cases

Database operations that must be sequential
State mutations that can't be concurrent
Processing events in order
Cache updates

Pattern: Natural backpressure (FIFO queue)

Single-Flight Processor

Deduplicates concurrent identical requests - only one in-flight operation per key.

Basic Usage

import io.getstream.android.core.api.processing.StreamSingleFlightProcessor

val singleFlight = StreamSingleFlightProcessor(scope)

// Multiple concurrent calls with same key share the same result
launch {
    val result = singleFlight.run("user-123") {
        // Expensive operation runs only once
        fetchUserFromNetwork("user-123")
    }
}

launch {
    val result = singleFlight.run("user-123") {
        // This waits for the first call, doesn't execute
        fetchUserFromNetwork("user-123")
    }
}

Use Cases

Token refresh (multiple requests trigger one refresh)
User data fetching
Configuration loading
Any expensive operation called concurrently

Pattern: First caller executes, subsequent callers await the same result

Retry Processor

Automatic retry with linear or exponential backoff.

Linear Backoff

import io.getstream.android.core.api.processing.StreamRetryProcessor
import io.getstream.android.core.api.model.StreamRetryPolicy

val retryProcessor = StreamRetryProcessor(logger)

val policy = StreamRetryPolicy.Linear(
    minRetries = 3,
    maxRetries = 10,
    initialDelayMillis = 1000, // 1s, 2s, 3s, ...
    maxDelayMillis = 30000      // Cap at 30s
)

val result = retryProcessor.retry(policy) {
    // Operation that may fail
    sendRequest()
}

Exponential Backoff

val policy = StreamRetryPolicy.Exponential(
    minRetries = 3,
    maxRetries = 10,
    initialDelayMillis = 1000, // 1s, 2s, 4s, 8s, 16s, ...
    maxDelayMillis = 60000,     // Cap at 60s
    giveUpFunction = { attempt, error ->
        // Stop retrying on specific errors
        error is UnauthorizedException
    }
)

Use Cases

Network requests with transient failures
WebSocket reconnection
Resource acquisition
Third-party API calls

Batcher

Collects items into batches based on size and/or time windows.

Basic Usage

import io.getstream.android.core.api.processing.StreamBatcher

val batcher = StreamBatcher<MyEvent>(
    scope = scope,
    batchSize = 10,           // Flush after 10 items
    initialDelayMs = 100,     // Initial wait before first flush
    maxDelayMs = 1000,        // Maximum time before forcing flush
    autoStart = true          // Start automatically on first enqueue
)

// Register batch handler
batcher.onBatch { batch, delayMs, count ->
    // Process batch
    sendBatchToServer(batch)
}

// Enqueue items
batcher.enqueue(event1)
batcher.enqueue(event2)
// Batched and sent together

batcher.stop().getOrThrow()

Use Cases

Analytics event batching
Log aggregation
Bulk API requests
Reducing network calls

Pattern: Debouncing + size-based flushing

Debouncer

Coalesces rapid state changes into a single settled action. Only the last value is delivered after a quiet period.

Basic Usage

import io.getstream.android.core.api.processing.StreamDebouncer

val debouncer = StreamDebouncer<String>(
    scope = scope,
    logger = logger,
    delayMs = 300 // Deliver after 300ms of silence
)

debouncer.onValue { settled ->
    // Only called once with the final value
    updateSearch(settled)
}

// Rapid submissions — only "third" is delivered after 300ms of silence
debouncer.submit("first")
debouncer.submit("second")
debouncer.submit("third")

// Cancel pending delivery
debouncer.cancel()

Use Cases

Search-as-you-type (wait for user to stop typing)
Network/lifecycle state coalescing (avoid reconnection storms)
UI state settling (wait for animations to finish)

Pattern: Last-write-wins with timer reset on each submission

Throttler

Rate-limits bursty values with configurable strategies.

Strategies

import io.getstream.android.core.api.processing.StreamThrottler
import io.getstream.android.core.api.processing.StreamThrottlePolicy

// Leading: first value immediately, drop the rest until window expires
val typing = StreamThrottler<TypingEvent>(
    scope = scope,
    logger = logger,
    policy = StreamThrottlePolicy.leading(windowMs = 3_000)
)

// Trailing: collect during window, deliver last value when it expires
val position = StreamThrottler<Position>(
    scope = scope,
    logger = logger,
    policy = StreamThrottlePolicy.trailing(windowMs = 1_000)
)

// Leading + Trailing: first value immediately, last value at window end
val scroll = StreamThrottler<Int>(
    scope = scope,
    logger = logger,
    policy = StreamThrottlePolicy.leadingAndTrailing(windowMs = 500)
)

Basic Usage

val throttler = StreamThrottler<String>(
    scope = scope,
    logger = logger,
    policy = StreamThrottlePolicy.leading(windowMs = 3_000)
)

throttler.onValue { event -> api.sendTypingEvent(event) }

throttler.submit("typing") // delivered immediately
throttler.submit("typing") // dropped (within window)
// ... 3 seconds pass ...
throttler.submit("typing") // delivered immediately (new window)

// Reset window manually
throttler.reset()

Use Cases

Typing indicators (leading — fire immediately, suppress duplicates)
Read receipts (leading — don't flood the server)
Position/progress updates (trailing — only latest state matters)
Scroll tracking (leading + trailing — responsiveness + final accuracy)

Pattern: Configurable rate-limiting via StreamThrottlePolicy

Event Aggregator

Adaptive event aggregator that switches between individual and batched event delivery based on traffic volume.

How It Works

During normal traffic, WebSocket events are deserialized and dispatched one at a time. During spikes, the aggregator collects events within a time window and delivers them as a single StreamAggregatedEvent — a flat list preserving arrival order.

import io.getstream.android.core.api.processing.StreamEventAggregator
import io.getstream.android.core.api.processing.StreamEventAggregationPolicy
import io.getstream.android.core.api.processing.StreamAggregatedEvent

val aggregator = StreamEventAggregator<MyEvent>(
    scope = scope,
    policy = StreamEventAggregationPolicy.from(
        typeExtractor = { raw -> extractType(raw) },
        deserializer = { raw -> Result.success(deserialize(raw)) },
        aggregationThreshold = 10,  // Aggregate when >= 10 events accumulated
        maxWindowMs = 500,          // Max collection window before flushing
    ),
)

aggregator.onEvent { event ->
    when (event) {
        is StreamAggregatedEvent<*> -> {
            // Spike — process all events sequentially in one atomic state update
            event.events.forEach { singleEvent ->
                handleEvent(singleEvent)
            }
        }
        is MyEvent -> {
            // Normal traffic — single event
            handleEvent(event)
        }
    }
}

aggregator.start()

Event Ordering

StreamAggregatedEvent.events is a flat List<T> in arrival order — not grouped by type. This is critical for correctness: events like "reaction added" must be processed before "reaction removed" for the same entity. Grouping by type key would lose this inter-type ordering.

Product SDKs should process the list sequentially and apply all updates in one atomic state mutation, so UI recomposes once per batch instead of once per event.

Handling Aggregated Events in Subscriptions

When subscribing to StreamClient events via StreamClientListener, your onEvent handler may receive either an individual event or a StreamAggregatedEvent. Product SDKs must handle both:

val listener = object : StreamClientListener {
    override fun onEvent(event: Any) {
        when (event) {
            is StreamAggregatedEvent<*> -> {
                // Apply all events in one state transaction
                val events = event.events.filterIsInstance<MyProductEvent>()
                stateStore.atomicUpdate { state ->
                    events.fold(state) { acc, e -> applyEvent(acc, e) }
                }
            }
            is MyProductEvent -> {
                // Single event — apply directly
                stateStore.update { state -> applyEvent(state, event) }
            }
        }
    }
}

Failing to handle StreamAggregatedEvent means aggregated events are silently ignored during traffic spikes.

Use Cases

WebSocket event delivery (automatic — built into StreamSocketSession)
High-frequency event streams (chat rooms with many participants)
Any scenario where burst traffic causes excessive UI recomposition

Threading Utilities

Safe cross-thread execution with timeout protection.

Running on Main Thread

import io.getstream.android.core.api.utils.runOnMainLooper

// From any thread, execute on main thread
val result = runOnMainLooper {
    // Runs on main thread
    lifecycle.addObserver(observer)
}

result
    .onSuccess { /* observer added */ }
    .onFailure { error -> logger.e(error) { "Failed to add observer" } }

Running on Custom Looper

import io.getstream.android.core.api.utils.runOn
import android.os.HandlerThread

val workerThread = HandlerThread("worker").apply { start() }
val workerLooper = workerThread.looper

val result = runOn(workerLooper) {
    // Runs on worker thread
    performBackgroundWork()
}

Key Features

Automatic thread detection: If already on target thread, executes immediately
Blocking with timeout: Caller blocks until completion (5 second timeout)
Exception propagation: Exceptions captured in Result
Cancellation-safe: CancellationException is rethrown

Pitfall: This is a blocking operation. For non-blocking alternatives, use coroutines with appropriate dispatchers:

// Non-blocking alternative
withContext(Dispatchers.Main) {
    lifecycle.addObserver(observer)
}

Token Management

Handles authentication token lifecycle with automatic refresh.

Basic Setup

import io.getstream.android.core.api.authentication.StreamTokenManager
import io.getstream.android.core.api.authentication.StreamTokenProvider

val tokenProvider = object : StreamTokenProvider {
    override suspend fun getToken(): String {
        // Fetch token from backend
        return api.getAuthToken(userId)
    }
}

val tokenManager = StreamTokenManager(
    userId = userId,
    tokenProvider = tokenProvider,
    singleFlight = singleFlight // Prevents multiple concurrent refreshes
)

// Get current token (triggers refresh if needed)
val token = tokenManager.getToken()

Token Refresh Flow

// Set initial token
tokenManager.setToken("initial-token")

// Token becomes invalid
tokenManager.invalidateToken()

// Next call triggers refresh via tokenProvider
val newToken = tokenManager.getToken() // Calls tokenProvider.getToken()

Key Points:

Token refresh is deduplicated via single-flight processor
Multiple concurrent requests wait for single refresh
Thread-safe token updates

WebSocket Connections

Reliable WebSocket connections with health monitoring.

Creating a WebSocket

import io.getstream.android.core.api.socket.StreamWebSocketFactory
import io.getstream.android.core.api.socket.listeners.StreamWebSocketListener

val socketFactory = StreamWebSocketFactory(logger)

val listener = object : StreamWebSocketListener {
    override fun onConnected(webSocket: StreamWebSocket) {
        logger.i { "WebSocket connected" }
    }

    override fun onMessage(webSocket: StreamWebSocket, text: String) {
        logger.d { "Received: $text" }
    }

    override fun onClosing(webSocket: StreamWebSocket, code: Int, reason: String) {
        logger.w { "Closing: $code - $reason" }
    }

    override fun onFailure(webSocket: StreamWebSocket, t: Throwable) {
        logger.e(t) { "WebSocket error" }
    }
}

val webSocket = socketFactory.create(
    url = "wss://example.com/socket",
    listener = listener
)

webSocket.connect()

Health Monitoring

import io.getstream.android.core.api.socket.monitor.StreamHealthMonitor

val healthMonitor = StreamHealthMonitor(
    logger = logger,
    scope = scope,
    pingInterval = 30_000, // Ping every 30s
    pongTimeout = 10_000   // Expect pong within 10s
)

healthMonitor.start().getOrThrow()

// Monitor will detect unhealthy connections

Logging

Structured, configurable logging throughout the library.

Setup

import io.getstream.android.core.api.log.StreamLoggerProvider
import io.getstream.android.core.api.log.StreamLogger

// Default Android logger
val logProvider = StreamLoggerProvider.defaultAndroidLogger(
    minLevel = StreamLogger.LogLevel.Debug,
    honorAndroidIsLoggable = true // Respects Log.isLoggable()
)

// Tagged logger for specific component
val logger = logProvider.taggedLogger("MyComponent")

Usage

logger.v { "Verbose message" }        // Verbose
logger.d { "Debug message" }          // Debug
logger.i { "Info message" }           // Info
logger.w { "Warning message" }        // Warn
logger.e(exception) { "Error" }       // Error with exception

// Lazy evaluation - only called if log level is enabled
logger.d {
    "Expensive: ${computeExpensiveString()}"
}

Log Levels

enum class LogLevel {
    VERBOSE,
    DEBUG,
    INFO,
    WARN,
    ERROR,
    NONE
}

Common Pitfalls & Best Practices

1. Lifecycle Threading Violations

Problem: Android Lifecycle components require operations on the main thread.

// ❌ WRONG - May crash with "Can only call from main thread"
thread {
    lifecycle.addObserver(observer)
}

Solution: Use runOnMainLooper:

// ✅ CORRECT
thread {
    runOnMainLooper {
        lifecycle.addObserver(observer)
    }.getOrThrow()
}

Why: The library uses runOnMainLooper internally in StreamLifecycleMonitorImpl to ensure thread safety.

2. Shared Subscription Managers

Problem: Reusing the same subscription manager instance can cause event loops.

// ❌ WRONG
val sharedManager = StreamSubscriptionManager<MyListener>(logger)

val client = StreamClient(
    subscriptionManager = sharedManager,
    // ...
)

val session = StreamSession(
    subscriptionManager = sharedManager, // Same instance!
    // ...
)

Solution: Create separate instances for different components:

// ✅ CORRECT
val clientManager = StreamSubscriptionManager<ClientListener>(logger)
val sessionManager = StreamSubscriptionManager<SessionListener>(logger)

Why: Shared managers can cause nested event notifications leading to stack overflows or infinite loops.

3. Forgetting to Start/Stop Components

Problem: Components don't work until started, resources leak if not stopped.

// ❌ WRONG
val queue = StreamSerialProcessingQueue(logger, scope)
queue.submit { /* Never runs! */ }

Solution: Always follow the start/use/stop pattern:

// ✅ CORRECT
val queue = StreamSerialProcessingQueue(logger, scope)
queue.start().getOrThrow()

queue.submit { /* Runs correctly */ }

queue.stop().getOrThrow()

4. Blocking Main Thread with runOn

Problem: runOn and runOnMainLooper block the calling thread.

// ❌ WRONG - Blocks main thread
fun onButtonClick() {
    runOn(backgroundLooper) {
        Thread.sleep(10_000) // Main thread blocked for 10s!
    }
}

Solution: Use coroutines for non-blocking cross-thread operations:

// ✅ CORRECT
fun onButtonClick() {
    lifecycleScope.launch {
        withContext(Dispatchers.IO) {
            performWork()
        }
    }
}

When to use runOn: Only when you need synchronous, guaranteed-ordered execution (e.g., lifecycle operations).

5. Ignoring Result Failures

Problem: Silent failures lead to hard-to-debug issues.

// ❌ WRONG
component.start() // Ignores potential failure

Solution: Always handle Result:

// ✅ CORRECT
component.start()
    .onSuccess { logger.i { "Started successfully" } }
    .onFailure { error ->
        logger.e(error) { "Failed to start component" }
        // Handle error appropriately
    }

6. Coroutine Scope Lifecycle Mismatches

Problem: Using wrong scope leads to leaks or premature cancellation.

// ❌ WRONG - GlobalScope never cancels
val queue = StreamSerialProcessingQueue(
    logger = logger,
    scope = GlobalScope
)

Solution: Use appropriately scoped coroutines:

// ✅ CORRECT
class MyComponent : ViewModel() {
    private val queue = StreamSerialProcessingQueue(
        logger = logger,
        scope = viewModelScope // Cancelled when ViewModel cleared
    )
}

Pattern: Match component lifetime to scope lifetime.

7. Race Conditions in Serial Queue

Problem: Submitting from multiple threads without synchronization.

// ❌ POTENTIALLY WRONG
thread { queue.submit { work1() } }
thread { queue.submit { work2() } }
// Order not guaranteed!

Solution: Queue operations are thread-safe, but if order matters across threads, use explicit synchronization:

// ✅ CORRECT - If order must be guaranteed
synchronized(lock) {
    queue.submit { work1() }
    queue.submit { work2() }
}

8. Token Refresh Storms

Problem: Multiple components triggering token refresh simultaneously.

// ❌ WRONG - Each component has its own token manager
val tokenManager1 = StreamTokenManager(userId, tokenProvider, singleFlight1)
val tokenManager2 = StreamTokenManager(userId, tokenProvider, singleFlight2)
// Both refresh tokens independently!

Solution: Share a single token manager instance:

// ✅ CORRECT
val tokenManager = StreamTokenManager(userId, tokenProvider, singleFlight)

val component1 = Component1(tokenManager)
val component2 = Component2(tokenManager)

Why: Single-flight processor in token manager ensures one refresh at a time.

9. Memory Leaks with Strong References

Problem: Listeners held with KEEP_UNTIL_CANCELLED but never cancelled.

// ❌ WRONG
class MyActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

        subscriptionManager.subscribe(
            listener = this,
            options = Options(retention = KEEP_UNTIL_CANCELLED)
        )
        // Never cancelled - activity leaks!
    }
}

Solution: Either use AUTO_REMOVE or cancel explicitly:

// ✅ CORRECT - Option 1: Auto cleanup
subscriptionManager.subscribe(
    listener = this,
    options = Options(retention = AUTO_REMOVE)
)

// ✅ CORRECT - Option 2: Manual cleanup
class MyActivity : AppCompatActivity() {
    private var subscription: StreamSubscription? = null

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        subscription = subscriptionManager.subscribe(
            listener = this,
            options = Options(retention = KEEP_UNTIL_CANCELLED)
        ).getOrThrow()
    }

    override fun onDestroy() {
        super.onDestroy()
        subscription?.cancel()
    }
}

10. Excessive Retry Attempts

Problem: Retrying operations that will never succeed.

// ❌ WRONG - Retries 401 Unauthorized forever
val policy = StreamRetryPolicy.Exponential(
    maxRetries = Int.MAX_VALUE,
    initialDelayMillis = 1000
)

retryProcessor.retry(policy) {
    authenticatedRequest() // Always returns 401
}

Solution: Use giveUpFunction to stop on non-retryable errors:

// ✅ CORRECT
val policy = StreamRetryPolicy.Exponential(
    maxRetries = 10,
    initialDelayMillis = 1000,
    giveUpFunction = { attempt, error ->
        when (error) {
            is UnauthorizedException,
            is ForbiddenException,
            is NotFoundException -> true // Don't retry
            else -> false // Continue retrying
        }
    }
)

Advanced Topics

Custom Implementations

All public interfaces can be replaced with custom implementations:

// Custom subscription manager
class MySubscriptionManager<T> : StreamSubscriptionManager<T> {
    override fun subscribe(listener: T, options: Options): Result<StreamSubscription> {
        // Custom logic
    }

    override fun clear() {
        // Custom cleanup
    }

    override fun forEach(action: (T) -> Unit): Result<Unit> {
        // Custom iteration
    }
}

Factory Functions

Prefer factory functions over direct instantiation:

// ✅ CORRECT - Factory function
val queue = StreamSerialProcessingQueue(logger, scope)

// ❌ AVOID - Direct instantiation
val queue = StreamSerialProcessingQueueImpl(logger, scope)

Why: Factory functions provide stability across implementation changes.

Testing

Mocking Components

Use factory functions with test implementations:

class TestSerialQueue : StreamSerialProcessingQueue {
    val submittedWork = mutableListOf<suspend () -> Any>()

    override suspend fun <T : Any> submit(job: suspend () -> T): Result<T> {
        submittedWork.add(job)
        @Suppress("UNCHECKED_CAST")
        return Result.success(Unit as T)
    }

    override suspend fun start(): Result<Unit> = Result.success(Unit)
    override suspend fun stop(timeout: Long?): Result<Unit> = Result.success(Unit)
}

@Test
fun testComponent() {
    val testQueue = TestSerialQueue()
    val component = MyComponent(testQueue)

    component.doSomething()

    assertEquals(1, testQueue.submittedWork.size)
}

Robolectric for Lifecycle Tests

Android lifecycle components require Robolectric:

@RunWith(RobolectricTestRunner::class)
@Config(sdk = [Build.VERSION_CODES.UPSIDE_DOWN_CAKE])
class LifecycleTest {
    @Test
    fun testLifecycleMonitor() {
        val monitor = StreamLifecycleMonitor(...)
        // Test with real Android components
    }
}

Project Structure

stream-android-core/ ├── app/ # Demo app for manual testing ├── stream-android-core/ # Core library │ └── src/main/java/io/getstream/android/core/ │ ├── api/ # Public API │ │ ├── authentication/ # Token management │ │ ├── observers/ # Lifecycle & network monitoring │ │ │ ├── lifecycle/ │ │ │ └── network/ │ │ ├── processing/ # Queue, retry, single-flight, batcher, debouncer, throttler │ │ ├── subscribe/ # Subscription management │ │ ├── socket/ # WebSocket connections │ │ ├── log/ # Logging infrastructure │ │ ├── utils/ # Threading, Result utilities │ │ └── model/ # Data models │ └── internal/ # Internal implementations ├── stream-android-core-annotations/ # Annotations & processors ├── stream-android-core-lint/ # Custom lint rules ├── config/ # Static analysis configs └── gradle/ # Gradle configuration

License

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-core-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.

About

Core SDK for the Android products.

Readme

View license

Activity

Custom properties

2 stars

1 watching

0 forks

Report repository

Releases 11

v5.0.1 Latest

Jul 7, 2026

+ 10 releases

Packages

Contributors

Languages

Kotlin 100.0%

Terms
Privacy
Security
Status
Community
Docs
Contact
Manage cookies
Do not share my personal information

Latest commit History 121 Commits
.github		.github
.planning		.planning
app		app
config		config
gradle		gradle
stream-android-core-annotations		stream-android-core-annotations
stream-android-core-lint		stream-android-core-lint
stream-android-core		stream-android-core
.gitignore		.gitignore
AGENTS.md		AGENTS.md
CLAUDE.md		CLAUDE.md
LICENSE		LICENSE
README.md		README.md
build.gradle.kts		build.gradle.kts
gradle.properties		gradle.properties
gradlew		gradlew
gradlew.bat		gradlew.bat
lint.xml		lint.xml
settings.gradle.kts		settings.gradle.kts
View all files

Navigation Menu

Search code, repositories, users, issues, pull requests...

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

GetStream/stream-core-android

Folders and files

Latest commit

History

Repository files navigation

Stream Android Core

What is Stream?

What is Stream Android Core?

Purpose

Architecture Position

Overview

API Visibility: Internal vs Published

Who Should Use This?

Table of Contents

Quick Start

Minimal Setup

Basic Client Instantiation

Requirements

Core Concepts

Component Lifecycle

Result-Based Error Handling

Thread Safety

Feature Guides

Lifecycle & Network Monitoring

Lifecycle Monitoring

Network Monitoring

Subscription Management

Basic Usage

Retention Policies

Handling Aggregated Events

Serial Processing Queue

Basic Usage

Use Cases

Single-Flight Processor

Basic Usage

Use Cases

Retry Processor

Linear Backoff

Exponential Backoff

Use Cases

Batcher

Basic Usage

Use Cases

Debouncer

Basic Usage

Use Cases

Throttler

Strategies

Basic Usage

Use Cases

Event Aggregator

How It Works

Event Ordering

Handling Aggregated Events in Subscriptions

Use Cases

Threading Utilities

Running on Main Thread

Running on Custom Looper

Key Features

Token Management

Basic Setup

Token Refresh Flow

WebSocket Connections

Creating a WebSocket

Health Monitoring

Logging

Setup

Usage

Log Levels

Common Pitfalls & Best Practices

1. Lifecycle Threading Violations

2. Shared Subscription Managers

3. Forgetting to Start/Stop Components

4. Blocking Main Thread with runOn

Packages