diff --git a/README.md b/README.md index 08fbb89..249a552 100644 --- a/README.md +++ b/README.md @@ -113,11 +113,27 @@ Stream Android Core uses annotations to distinguish between stable public APIs a ## Quick Start +> **Before you copy anything below:** nearly every type in this library is annotated +> `@StreamInternalApi`, whose opt-in level is `ERROR` — including `StreamClient` itself, +> `StreamUser`, `StreamSocketConfig`, `StreamLogger`, and the processing utilities. Only a handful +> of value types (`StreamApiKey`, `StreamUserId`, `StreamToken`) are `@StreamPublishedApi`. Add a +> file-level opt-in once and every snippet on this page compiles: +> +> ```kotlin +> @file:OptIn(StreamInternalApi::class) +> +> import io.getstream.android.core.annotations.StreamInternalApi +> ``` +> +> A few snippets below still show `@OptIn(...)` inline where it is worth calling out; the +> file-level form above covers all of them. + ### Minimal Setup Here's a minimal example to get started with Stream Android Core: ```kotlin +import io.getstream.android.core.api.log.StreamLogger import io.getstream.android.core.api.log.StreamLoggerProvider import io.getstream.android.core.api.subscribe.StreamSubscriptionManager import kotlinx.coroutines.CoroutineScope @@ -141,25 +157,54 @@ val subscriptionManager = StreamSubscriptionManager( ### Basic Client Instantiation ```kotlin -val singleFlight = StreamSingleFlightProcessor(scope) -val tokenManager = StreamTokenManager(userId, tokenProvider, singleFlight) - -val serialQueue = StreamSerialProcessingQueue( - logger = logProvider.taggedLogger("SerialQueue"), - scope = scope -) +import android.os.Build +import io.getstream.android.core.api.StreamClient +import io.getstream.android.core.api.authentication.StreamTokenProvider +import io.getstream.android.core.api.model.StreamUser +import io.getstream.android.core.api.model.config.StreamClientSerializationConfig +import io.getstream.android.core.api.model.config.StreamSocketConfig +import io.getstream.android.core.api.model.value.StreamApiKey +import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader +import io.getstream.android.core.api.model.value.StreamToken +import io.getstream.android.core.api.model.value.StreamUserId +import io.getstream.android.core.api.model.value.StreamWsUrl +import io.getstream.android.core.api.serialization.StreamEventSerialization + +val user = StreamUser(StreamUserId.fromString("sample-user")) +val token = StreamToken.fromString("") val client = StreamClient( scope = scope, - apiKey = apiKey, - userId = userId, - tokenProvider = tokenProvider, - logProvider = logProvider, - tokenManager = tokenManager, - singleFlight = singleFlight, - serialQueue = serialQueue, - // ... other dependencies + context = context.applicationContext, + user = user, + products = listOf("chat"), // e.g. listOf("chat", "video", "feeds") + tokenProvider = object : StreamTokenProvider { + override suspend fun loadToken(userId: StreamUserId): StreamToken = token + }, + serializationConfig = StreamClientSerializationConfig.default( + // Product SDKs supply their own event (de)serializer; Unit shown for brevity + object : StreamEventSerialization { + override fun serialize(data: Unit): Result = Result.success("") + override fun deserialize(raw: String): Result = Result.success(Unit) + } + ), + socketConfig = StreamSocketConfig.jwt( + url = StreamWsUrl.fromString("wss://.stream-io-api.com/api/v2/connect"), + apiKey = StreamApiKey.fromString(""), + clientInfoHeader = StreamHttpClientInfoHeader.create( + product = "android-core", + productVersion = "1.1.0", + os = "Android", + apiLevel = Build.VERSION.SDK_INT, + deviceModel = "Pixel 7 Pro", + app = "My App", + appVersion = "1.0.0", + ), + ), ) + +// Open the socket, authenticate, and start monitoring +val connected = client.connect() // Result ``` --- @@ -240,8 +285,10 @@ Monitor app lifecycle (foreground/background) and network connectivity changes. #### Lifecycle Monitoring ```kotlin -import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor +import io.getstream.android.core.api.model.connection.lifecycle.StreamLifecycleState import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleListener +import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor +import io.getstream.android.core.api.subscribe.StreamSubscriptionManager // Create the monitor val lifecycleMonitor = StreamLifecycleMonitor( @@ -278,8 +325,10 @@ when (state) { #### Network Monitoring ```kotlin +import io.getstream.android.core.api.model.connection.network.StreamNetworkInfo import io.getstream.android.core.api.observers.network.StreamNetworkMonitor import io.getstream.android.core.api.observers.network.StreamNetworkMonitorListener +import io.getstream.android.core.api.subscribe.StreamSubscriptionManager val networkMonitor = StreamNetworkMonitor( context = context, @@ -432,19 +481,23 @@ Deduplicates concurrent identical requests - only one in-flight operation per ke ```kotlin import io.getstream.android.core.api.processing.StreamSingleFlightProcessor +import io.getstream.android.core.api.model.StreamTypedKey val singleFlight = StreamSingleFlightProcessor(scope) -// Multiple concurrent calls with same key share the same result +// Keys are StreamTypedKey — equal keys (by id) share the same in-flight result +val userKey = StreamTypedKey("user-123") + +// Multiple concurrent calls with the same key share the same result launch { - val result = singleFlight.run("user-123") { + val result = singleFlight.run(userKey) { // Expensive operation runs only once fetchUserFromNetwork("user-123") } } launch { - val result = singleFlight.run("user-123") { + val result = singleFlight.run(userKey) { // This waits for the first call, doesn't execute fetchUserFromNetwork("user-123") } @@ -464,21 +517,42 @@ launch { ### Retry Processor -Automatic retry with linear or exponential backoff. +Automatic retry with a choice of back-off curves. + +> **Note:** `StreamRetryPolicy` is annotated `@StreamInternalApi` — using it requires +> `@OptIn(StreamInternalApi::class)`. Policies are created via the `linear` / `quadratic` / +> `exponential` / `fixed` / `custom` factory methods (the primary constructor is private). + +Delays for the first four retries at `backoffStepMillis = 250` (`fixed` shown at its own +`delayMillis = 500` default): + +| Retry | `linear` | `exponential` | `quadratic` | `fixed` | +|---|---|---|---|---| +| 1 | 250 ms | 250 ms | 250 ms | 500 ms | +| 2 | 500 ms | 500 ms | 750 ms | 500 ms | +| 3 | 750 ms | 1 000 ms | 1 500 ms | 500 ms | +| 4 | 1 000 ms | 2 000 ms | 2 500 ms | 500 ms | + +The table stops at four deliberately: the processor gives up on the last attempt *before* asking +the policy for another delay, so with the default `maxRetries = 5` a fifth delay is never +computed. That matters for `quadratic` in particular — doubling only overtakes it at retry 5, so +in practice `quadratic` always waits longer than `exponential`, not less. #### Linear Backoff ```kotlin +import io.getstream.android.core.annotations.StreamInternalApi import io.getstream.android.core.api.processing.StreamRetryProcessor -import io.getstream.android.core.api.model.StreamRetryPolicy +import io.getstream.android.core.api.model.retry.StreamRetryPolicy val retryProcessor = StreamRetryProcessor(logger) -val policy = StreamRetryPolicy.Linear( +@OptIn(StreamInternalApi::class) +val policy = StreamRetryPolicy.linear( minRetries = 3, maxRetries = 10, - initialDelayMillis = 1000, // 1s, 2s, 3s, ... - maxDelayMillis = 30000 // Cap at 30s + backoffStepMillis = 250, // 250ms, 500ms, 750ms, ... + maxBackoffMillis = 30_000, // Cap at 30s ) val result = retryProcessor.retry(policy) { @@ -490,18 +564,76 @@ val result = retryProcessor.retry(policy) { #### Exponential Backoff ```kotlin -val policy = StreamRetryPolicy.Exponential( +@OptIn(StreamInternalApi::class) +val policy = StreamRetryPolicy.exponential( minRetries = 3, maxRetries = 10, - initialDelayMillis = 1000, // 1s, 2s, 4s, 8s, 16s, ... - maxDelayMillis = 60000, // Cap at 60s - giveUpFunction = { attempt, error -> + backoffStepMillis = 250, // grows: 250ms, 500ms, 1s, 2s, ... + maxBackoffMillis = 60_000, // Cap at 60s + giveUp = { retry, error -> // Stop retrying on specific errors error is UnauthorizedException } ) ``` +#### Quadratic Backoff + +Adds a growing increment to the previous delay, so the waits follow the triangular numbers scaled +by the step (`backoffStepMillis × n(n + 1) / 2`). + +```kotlin +@OptIn(StreamInternalApi::class) +val policy = StreamRetryPolicy.quadratic( + minRetries = 3, + maxRetries = 10, + backoffStepMillis = 250, // grows: 250ms, 750ms, 1.5s, 2.5s, ... + maxBackoffMillis = 60_000, // Cap at 60s +) +``` + +Unlike `exponential`, this curve reads the previous delay, so it only behaves as described when +delays are fed back in — which `StreamRetryProcessor` does. + +#### Fixed Delay + +Every retry waits the same amount. + +```kotlin +@OptIn(StreamInternalApi::class) +val policy = StreamRetryPolicy.fixed( + minRetries = 1, + maxRetries = 5, + delayMillis = 500, // 500ms before every retry + maxBackoffMillis = 15_000, +) +``` + +#### Custom Backoff + +Supply the curve yourself. Every parameter is explicit, and the result is still clamped to +`[minBackoffMills, maxBackoffMills]` and validated against the same invariants as the built-ins. + +```kotlin +import io.getstream.android.core.annotations.StreamInternalApi +import io.getstream.android.core.api.model.retry.StreamRetryPolicy +import kotlin.random.Random + +@OptIn(StreamInternalApi::class) +val jitterPolicy = StreamRetryPolicy.custom( + minRetries = 2, + maxRetries = 6, + minBackoffMills = 200, + maxBackoffMills = 10_000, + initialDelayMillis = 100, + giveUp = { _, cause -> cause is UnauthorizedException }, +) { retry, previousDelay -> + // 1-based retry index, previous delay in ms — add ±20% jitter to a doubling curve + val base = previousDelay * 2 + (base * Random.nextDouble(0.8, 1.2)).toLong() +} +``` + #### Use Cases - Network requests with transient failures @@ -600,7 +732,7 @@ Rate-limits bursty values with configurable strategies. ```kotlin import io.getstream.android.core.api.processing.StreamThrottler -import io.getstream.android.core.api.processing.StreamThrottlePolicy +import io.getstream.android.core.api.model.processing.StreamThrottlePolicy // Leading: first value immediately, drop the rest until window expires val typing = StreamThrottler( @@ -798,35 +930,35 @@ Handles authentication token lifecycle with automatic refresh. ```kotlin import io.getstream.android.core.api.authentication.StreamTokenManager import io.getstream.android.core.api.authentication.StreamTokenProvider +import io.getstream.android.core.api.model.value.StreamToken +import io.getstream.android.core.api.model.value.StreamUserId -val tokenProvider = object : StreamTokenProvider { - override suspend fun getToken(): String { - // Fetch token from backend - return api.getAuthToken(userId) - } +val tokenProvider = StreamTokenProvider { userId: StreamUserId -> + // Fetch token from backend, wrap in a StreamToken + StreamToken.fromString(api.getAuthToken(userId.rawValue)) } val tokenManager = StreamTokenManager( - userId = userId, + userId = StreamUserId.fromString("user-123"), tokenProvider = tokenProvider, singleFlight = singleFlight // Prevents multiple concurrent refreshes ) -// Get current token (triggers refresh if needed) -val token = tokenManager.getToken() +// Load the token, fetching via the provider only if absent (triggers refresh if needed) +val token = tokenManager.loadIfAbsent() ``` #### Token Refresh Flow ```kotlin // Set initial token -tokenManager.setToken("initial-token") +tokenManager.setToken(StreamToken.fromString("initial-token")) // Token becomes invalid -tokenManager.invalidateToken() +tokenManager.invalidate() -// Next call triggers refresh via tokenProvider -val newToken = tokenManager.getToken() // Calls tokenProvider.getToken() +// Force a refresh via tokenProvider.loadToken(...) +val newToken = tokenManager.refresh() ``` **Key Points**: @@ -843,6 +975,7 @@ Reliable WebSocket connections with health monitoring. #### Creating a WebSocket ```kotlin +import io.getstream.android.core.api.socket.StreamWebSocket import io.getstream.android.core.api.socket.StreamWebSocketFactory import io.getstream.android.core.api.socket.listeners.StreamWebSocketListener @@ -930,17 +1063,24 @@ logger.d { #### Log Levels +`LogLevel` is a sealed class of severity objects nested inside `StreamLogger` (higher `level` = +more severe): + ```kotlin -enum class LogLevel { - VERBOSE, - DEBUG, - INFO, - WARN, - ERROR, - NONE +interface StreamLogger { + sealed class LogLevel(val level: Int) { + object Verbose : LogLevel(1) + object Debug : LogLevel(2) + object Info : LogLevel(3) + object Warning : LogLevel(4) + object Error : LogLevel(5) + } } ``` +Because it is nested, always reference it through the outer type — `StreamLogger.LogLevel.Verbose`, +`StreamLogger.LogLevel.Debug`, and so on. + --- ## Common Pitfalls & Best Practices @@ -1210,9 +1350,10 @@ class MyActivity : AppCompatActivity() { ```kotlin // ❌ WRONG - Retries 401 Unauthorized forever -val policy = StreamRetryPolicy.Exponential( +@OptIn(StreamInternalApi::class) +val policy = StreamRetryPolicy.exponential( maxRetries = Int.MAX_VALUE, - initialDelayMillis = 1000 + backoffStepMillis = 1000 ) retryProcessor.retry(policy) { @@ -1220,14 +1361,15 @@ retryProcessor.retry(policy) { } ``` -**Solution**: Use `giveUpFunction` to stop on non-retryable errors: +**Solution**: Use `giveUp` to stop on non-retryable errors: ```kotlin // ✅ CORRECT -val policy = StreamRetryPolicy.Exponential( +@OptIn(StreamInternalApi::class) +val policy = StreamRetryPolicy.exponential( maxRetries = 10, - initialDelayMillis = 1000, - giveUpFunction = { attempt, error -> + backoffStepMillis = 1000, + giveUp = { attempt, error -> when (error) { is UnauthorizedException, is ForbiddenException, @@ -1357,7 +1499,7 @@ stream-android-core/ ## License -Copyright (c) 2014-2025 Stream.io Inc. All rights reserved. +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.