update claude.md
This commit is contained in:
nrobi144
@@ -0,0 +1,309 @@
|
||||
# Advanced Flow Operators
|
||||
|
||||
Comprehensive guide to Flow operators for complex async patterns in Amethyst.
|
||||
|
||||
## Transformation Operators
|
||||
|
||||
### flatMapLatest - Cancel Previous, Switch to New
|
||||
|
||||
**Use when:** Latest value matters, previous operations should cancel
|
||||
|
||||
```kotlin
|
||||
// User types in search box → cancel previous search
|
||||
searchQuery
|
||||
.flatMapLatest { query ->
|
||||
repository.search(query) // Cancels previous search
|
||||
}
|
||||
.collect { results -> updateUI(results) }
|
||||
```
|
||||
|
||||
**Amethyst pattern:**
|
||||
```kotlin
|
||||
// Switch relays based on latest account
|
||||
accountFlow
|
||||
.flatMapLatest { account ->
|
||||
relayPool.observeEvents(account.relays)
|
||||
}
|
||||
```
|
||||
|
||||
### flatMapConcat - Sequential Processing
|
||||
|
||||
**Use when:** Order matters, process one at a time
|
||||
|
||||
```kotlin
|
||||
eventIds
|
||||
.flatMapConcat { id ->
|
||||
repository.fetchEvent(id)
|
||||
}
|
||||
.collect { event -> process(event) }
|
||||
```
|
||||
|
||||
### flatMapMerge - Concurrent Processing
|
||||
|
||||
**Use when:** Process multiple simultaneously, order doesn't matter
|
||||
|
||||
```kotlin
|
||||
relays
|
||||
.flatMapMerge(concurrency = 10) { relay ->
|
||||
relay.subscribe(filters)
|
||||
}
|
||||
.collect { event -> handleEvent(event) }
|
||||
```
|
||||
|
||||
## Combination Operators
|
||||
|
||||
### combine - Latest from Multiple Flows
|
||||
|
||||
**Use when:** Need latest value from ALL flows
|
||||
|
||||
```kotlin
|
||||
combine(
|
||||
accountFlow,
|
||||
settingsFlow,
|
||||
connectivityFlow
|
||||
) { account, settings, connectivity ->
|
||||
AppState(account, settings, connectivity)
|
||||
}.collect { state -> render(state) }
|
||||
```
|
||||
|
||||
**Pattern:** Re-emits whenever ANY source emits
|
||||
|
||||
### zip - Pair Values in Order
|
||||
|
||||
**Use when:** Need corresponding values from flows
|
||||
|
||||
```kotlin
|
||||
zip(requestFlow, responseFlow) { req, res ->
|
||||
Pair(req, res)
|
||||
}
|
||||
```
|
||||
|
||||
**Pattern:** Waits for BOTH to emit before pairing
|
||||
|
||||
### merge - Combine Multiple Flows
|
||||
|
||||
**Use when:** Treat multiple flows as single stream
|
||||
|
||||
```kotlin
|
||||
merge(
|
||||
relay1.events,
|
||||
relay2.events,
|
||||
relay3.events
|
||||
).collect { event -> handleEvent(event) }
|
||||
```
|
||||
|
||||
## Backpressure & Buffering
|
||||
|
||||
### shareIn - Hot Flow from Cold
|
||||
|
||||
**Use when:** Multiple collectors should share single upstream
|
||||
|
||||
```kotlin
|
||||
val sharedEvents = repository.observeEvents()
|
||||
.shareIn(
|
||||
scope = viewModelScope,
|
||||
started = SharingStarted.WhileSubscribed(5000),
|
||||
replay = 0
|
||||
)
|
||||
|
||||
// Multiple collectors share same upstream
|
||||
sharedEvents.collect { /* collector 1 */ }
|
||||
sharedEvents.collect { /* collector 2 */ }
|
||||
```
|
||||
|
||||
**SharingStarted strategies:**
|
||||
- `Eagerly` - Start immediately, never stop
|
||||
- `Lazily` - Start on first subscriber, never stop
|
||||
- `WhileSubscribed(stopTimeout)` - Stop after last unsubscribe + timeout
|
||||
|
||||
### stateIn - StateFlow from Cold Flow
|
||||
|
||||
**Use when:** Convert Flow to StateFlow (always has value)
|
||||
|
||||
```kotlin
|
||||
val uiState: StateFlow<UiState> = repository.observeData()
|
||||
.map { data -> UiState.Success(data) }
|
||||
.stateIn(
|
||||
scope = viewModelScope,
|
||||
started = SharingStarted.WhileSubscribed(5000),
|
||||
initialValue = UiState.Loading
|
||||
)
|
||||
```
|
||||
|
||||
**Amethyst pattern:**
|
||||
```kotlin
|
||||
// Connectivity status as StateFlow
|
||||
val connectivity: StateFlow<ConnectivityStatus> =
|
||||
connectivityFlow.status
|
||||
.stateIn(
|
||||
scope = serviceScope,
|
||||
started = SharingStarted.Eagerly,
|
||||
initialValue = ConnectivityStatus.Off
|
||||
)
|
||||
```
|
||||
|
||||
### buffer - Control Backpressure
|
||||
|
||||
**Use when:** Producer faster than consumer
|
||||
|
||||
```kotlin
|
||||
eventFlow
|
||||
.buffer(capacity = 64, onBufferOverflow = BufferOverflow.DROP_OLDEST)
|
||||
.collect { event -> slowProcessor(event) }
|
||||
```
|
||||
|
||||
**Strategies:**
|
||||
- `SUSPEND` - Slow down producer (default)
|
||||
- `DROP_OLDEST` - Drop oldest in buffer
|
||||
- `DROP_LATEST` - Drop newest emission
|
||||
|
||||
### conflate - Keep Only Latest
|
||||
|
||||
**Use when:** Only latest value matters, skip intermediate
|
||||
|
||||
```kotlin
|
||||
locationFlow
|
||||
.conflate() // Skip intermediate locations
|
||||
.collect { location -> updateMap(location) }
|
||||
```
|
||||
|
||||
## Debouncing & Throttling
|
||||
|
||||
### debounce - Wait for Quiet Period
|
||||
|
||||
**Use when:** Wait for user to stop typing
|
||||
|
||||
```kotlin
|
||||
searchQuery
|
||||
.debounce(300) // Wait 300ms after last emission
|
||||
.flatMapLatest { query -> search(query) }
|
||||
```
|
||||
|
||||
**Amethyst pattern:**
|
||||
```kotlin
|
||||
// ConnectivityFlow.kt:87
|
||||
connectivityFlow
|
||||
.distinctUntilChanged()
|
||||
.debounce(200) // Wait 200ms for network to stabilize
|
||||
.flowOn(Dispatchers.IO)
|
||||
```
|
||||
|
||||
### sample - Periodic Sampling
|
||||
|
||||
**Use when:** Rate-limit high-frequency emissions
|
||||
|
||||
```kotlin
|
||||
sensorData
|
||||
.sample(1000) // Sample every 1 second
|
||||
.collect { data -> process(data) }
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
### catch - Handle Upstream Errors
|
||||
|
||||
**Use when:** Graceful degradation needed
|
||||
|
||||
```kotlin
|
||||
repository.fetchData()
|
||||
.catch { e ->
|
||||
Log.e("Error", e)
|
||||
emit(emptyList()) // Fallback value
|
||||
}
|
||||
.collect { data -> updateUI(data) }
|
||||
```
|
||||
|
||||
**Pattern:** Only catches UPSTREAM errors, not in collect block
|
||||
|
||||
### retry/retryWhen - Automatic Retry
|
||||
|
||||
```kotlin
|
||||
relayConnection
|
||||
.retry(3) { cause ->
|
||||
cause is IOException // Only retry on network errors
|
||||
}
|
||||
```
|
||||
|
||||
## Context Switching
|
||||
|
||||
### flowOn - Change Upstream Dispatcher
|
||||
|
||||
**Use when:** Offload work from current context
|
||||
|
||||
```kotlin
|
||||
repository.fetchData()
|
||||
.map { heavyProcessing(it) }
|
||||
.flowOn(Dispatchers.Default) // Heavy work on Default
|
||||
.collect { updateUI(it) } // Collect on Main
|
||||
```
|
||||
|
||||
**Critical:** Only affects UPSTREAM operators
|
||||
|
||||
**Amethyst pattern:**
|
||||
```kotlin
|
||||
// ConnectivityFlow.kt:87
|
||||
callbackFlow { /* ... */ }
|
||||
.distinctUntilChanged()
|
||||
.debounce(200)
|
||||
.flowOn(Dispatchers.IO) // All upstream on IO
|
||||
```
|
||||
|
||||
## Common Patterns
|
||||
|
||||
### Pattern: Multi-Relay Subscription
|
||||
|
||||
```kotlin
|
||||
fun observeFromMultipleRelays(relays: List<Relay>, filters: List<Filter>): Flow<Event> =
|
||||
relays.map { relay ->
|
||||
relay.subscribe(filters)
|
||||
}.merge()
|
||||
.distinctBy { it.id }
|
||||
```
|
||||
|
||||
### Pattern: Load + Cache + Observe
|
||||
|
||||
```kotlin
|
||||
fun observeWithCache(id: String): Flow<Data> = flow {
|
||||
// Emit cached value immediately
|
||||
cache[id]?.let { emit(it) }
|
||||
|
||||
// Then observe updates
|
||||
emitAll(repository.observe(id))
|
||||
}.distinctUntilChanged()
|
||||
```
|
||||
|
||||
### Pattern: Retry with Exponential Backoff
|
||||
|
||||
```kotlin
|
||||
fun <T> Flow<T>.retryWithBackoff(
|
||||
maxRetries: Int = 3,
|
||||
initialDelay: Long = 1000
|
||||
): Flow<T> = retryWhen { cause, attempt ->
|
||||
if (attempt >= maxRetries || cause !is IOException) {
|
||||
false
|
||||
} else {
|
||||
delay(initialDelay * (1L shl attempt.toInt()))
|
||||
true
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Performance Tips
|
||||
|
||||
1. **Use shareIn for expensive operations**
|
||||
- Compute once, share with multiple collectors
|
||||
|
||||
2. **Choose right backpressure strategy**
|
||||
- UI updates: `conflate()` or `DROP_OLDEST`
|
||||
- Events: `buffer()` with appropriate size
|
||||
|
||||
3. **flowOn placement matters**
|
||||
- Place after expensive operators to offload them
|
||||
|
||||
4. **Avoid unnecessary emissions**
|
||||
- Use `distinctUntilChanged()` when appropriate
|
||||
- Consider `debounce()` for high-frequency sources
|
||||
|
||||
5. **StateFlow vs SharedFlow**
|
||||
- StateFlow: Always has value, conflates
|
||||
- SharedFlow: Optional replay, configurable buffering
|
||||
@@ -0,0 +1,480 @@
|
||||
# Nostr Relay Async Patterns
|
||||
|
||||
Proven coroutine patterns for Nostr relay connections, subscriptions, and event streaming in Amethyst.
|
||||
|
||||
## Core Pattern: callbackFlow for Relay Subscriptions
|
||||
|
||||
### Pattern: Subscription as Flow
|
||||
|
||||
**Real implementation from NostrClientStaticReqAsStateFlow.kt:**
|
||||
|
||||
```kotlin
|
||||
fun INostrClient.reqAsFlow(
|
||||
relay: NormalizedRelayUrl,
|
||||
filters: List<Filter>,
|
||||
): Flow<List<Event>> =
|
||||
callbackFlow {
|
||||
val subId = RandomInstance.randomChars(10)
|
||||
var hasBeenLive = false
|
||||
val eventIds = mutableSetOf<HexKey>()
|
||||
var currentEvents = listOf<Event>()
|
||||
|
||||
val listener = object : IRequestListener {
|
||||
override fun onEvent(
|
||||
event: Event,
|
||||
isLive: Boolean,
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
if (event.id !in eventIds) {
|
||||
if (hasBeenLive) {
|
||||
// After EOSE: prepend new events
|
||||
val list = ArrayList<Event>(1 + currentEvents.size)
|
||||
list.add(event)
|
||||
list.addAll(currentEvents)
|
||||
currentEvents = list
|
||||
} else {
|
||||
// Before EOSE: append events
|
||||
currentEvents = currentEvents + event
|
||||
}
|
||||
eventIds.add(event.id)
|
||||
trySend(currentEvents)
|
||||
}
|
||||
}
|
||||
|
||||
override fun onEose(
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
hasBeenLive = true
|
||||
}
|
||||
}
|
||||
|
||||
openReqSubscription(subId, mapOf(relay to filters), listener)
|
||||
|
||||
awaitClose {
|
||||
close(subId)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Key techniques:**
|
||||
1. **callbackFlow** - Bridge callback API to Flow
|
||||
2. **Deduplication** - `eventIds` set prevents duplicates
|
||||
3. **EOSE handling** - Changes insertion strategy (append → prepend)
|
||||
4. **awaitClose** - Cleanup when flow cancelled
|
||||
5. **trySend** - Non-blocking emission from callback
|
||||
|
||||
## Multi-Relay Patterns
|
||||
|
||||
### Pattern: Merge Events from Multiple Relays
|
||||
|
||||
```kotlin
|
||||
fun observeFromRelays(
|
||||
relays: List<NormalizedRelayUrl>,
|
||||
filters: List<Filter>
|
||||
): Flow<Event> =
|
||||
relays.map { relay ->
|
||||
client.reqAsFlow(relay, filters)
|
||||
.flatMapConcat { it.asFlow() }
|
||||
}.merge()
|
||||
.distinctBy { it.id }
|
||||
```
|
||||
|
||||
**Explanation:**
|
||||
- Each relay produces `Flow<List<Event>>`
|
||||
- `flatMapConcat` flattens to `Flow<Event>`
|
||||
- `merge()` combines all relay flows
|
||||
- `distinctBy` deduplicates across relays
|
||||
|
||||
### Pattern: Concurrent Relay Operations with supervisorScope
|
||||
|
||||
```kotlin
|
||||
suspend fun subscribeToRelays(
|
||||
relays: List<Relay>,
|
||||
filters: List<Filter>
|
||||
) = supervisorScope {
|
||||
relays.forEach { relay ->
|
||||
launch {
|
||||
relay.subscribe(filters).collect { event ->
|
||||
eventChannel.send(event)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Why supervisorScope:**
|
||||
- If one relay fails, others continue
|
||||
- All children cancelled when scope cancelled
|
||||
- Structured concurrency maintained
|
||||
|
||||
## Backpressure Handling
|
||||
|
||||
### Pattern: Buffer with Drop Strategy
|
||||
|
||||
**For high-frequency event streams:**
|
||||
|
||||
```kotlin
|
||||
relayFlow
|
||||
.buffer(
|
||||
capacity = 64,
|
||||
onBufferOverflow = BufferOverflow.DROP_OLDEST
|
||||
)
|
||||
.collect { event -> processEvent(event) }
|
||||
```
|
||||
|
||||
**Strategy selection:**
|
||||
- `DROP_OLDEST` - For real-time feeds (lose old events OK)
|
||||
- `DROP_LATEST` - For priority queues (lose new events OK)
|
||||
- `SUSPEND` - For critical events (slow down producer)
|
||||
|
||||
### Pattern: Conflate for UI Updates
|
||||
|
||||
```kotlin
|
||||
val uiEvents: Flow<UiEvent> = relayEvents
|
||||
.map { event -> toUiEvent(event) }
|
||||
.conflate() // Skip intermediate, show latest
|
||||
.flowOn(Dispatchers.Default)
|
||||
```
|
||||
|
||||
## Connection Management
|
||||
|
||||
### Pattern: Network Connectivity as Flow
|
||||
|
||||
**Real implementation from ConnectivityFlow.kt:**
|
||||
|
||||
```kotlin
|
||||
@OptIn(FlowPreview::class)
|
||||
val status = callbackFlow {
|
||||
trySend(ConnectivityStatus.StartingService)
|
||||
|
||||
val connectivityManager = context.getConnectivityManager()
|
||||
|
||||
val networkCallback = object : ConnectivityManager.NetworkCallback() {
|
||||
override fun onAvailable(network: Network) {
|
||||
connectivityManager.getNetworkCapabilities(network)?.let {
|
||||
trySend(ConnectivityStatus.Active(
|
||||
network.networkHandle,
|
||||
it.isMeteredOrMobileData()
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
override fun onCapabilitiesChanged(
|
||||
network: Network,
|
||||
networkCapabilities: NetworkCapabilities
|
||||
) {
|
||||
val isMobile = networkCapabilities.isMeteredOrMobileData()
|
||||
trySend(ConnectivityStatus.Active(
|
||||
network.networkHandle,
|
||||
isMobile
|
||||
))
|
||||
}
|
||||
|
||||
override fun onLost(network: Network) {
|
||||
trySend(ConnectivityStatus.Off)
|
||||
}
|
||||
}
|
||||
|
||||
connectivityManager.registerDefaultNetworkCallback(networkCallback)
|
||||
|
||||
// Send initial state
|
||||
connectivityManager.activeNetwork?.let { network ->
|
||||
connectivityManager.getNetworkCapabilities(network)?.let {
|
||||
trySend(ConnectivityStatus.Active(
|
||||
network.networkHandle,
|
||||
it.isMeteredOrMobileData()
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
awaitClose {
|
||||
connectivityManager.unregisterNetworkCallback(networkCallback)
|
||||
trySend(ConnectivityStatus.Off)
|
||||
}
|
||||
}
|
||||
.distinctUntilChanged()
|
||||
.debounce(200) // Stabilize rapid changes
|
||||
.flowOn(Dispatchers.IO)
|
||||
```
|
||||
|
||||
**Key patterns:**
|
||||
1. **Initial state** - Emit current connectivity immediately
|
||||
2. **Callback registration** - Register listener in flow body
|
||||
3. **Cleanup** - Unregister in `awaitClose`
|
||||
4. **Stabilization** - `debounce(200)` prevents flapping
|
||||
5. **Deduplication** - `distinctUntilChanged()` skips redundant updates
|
||||
|
||||
### Pattern: Reconnect on Connectivity Change
|
||||
|
||||
```kotlin
|
||||
connectivityFlow
|
||||
.flatMapLatest { status ->
|
||||
when (status) {
|
||||
is ConnectivityStatus.Active -> {
|
||||
relayPool.connectAll()
|
||||
relayPool.observeEvents()
|
||||
}
|
||||
else -> emptyFlow()
|
||||
}
|
||||
}
|
||||
.collect { event -> handleEvent(event) }
|
||||
```
|
||||
|
||||
## Exception Handling in Async Operations
|
||||
|
||||
### Pattern: CoroutineExceptionHandler + SupervisorJob
|
||||
|
||||
**Real implementation from PushNotificationReceiverService.kt:**
|
||||
|
||||
```kotlin
|
||||
class PushNotificationReceiverService : FirebaseMessagingService() {
|
||||
// Catch all uncaught exceptions
|
||||
val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
|
||||
Log.e("AmethystCoroutine", "Caught exception: ${throwable.message}", throwable)
|
||||
}
|
||||
|
||||
// Children fail independently, handler catches all
|
||||
private val scope = CoroutineScope(
|
||||
Dispatchers.IO + SupervisorJob() + exceptionHandler
|
||||
)
|
||||
|
||||
override fun onMessageReceived(remoteMessage: RemoteMessage) {
|
||||
scope.launch(Dispatchers.IO) {
|
||||
parseMessage(remoteMessage.data)?.let { receiveIfNew(it) }
|
||||
}
|
||||
}
|
||||
|
||||
override fun onDestroy() {
|
||||
scope.cancel()
|
||||
super.onDestroy()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Why this pattern:**
|
||||
- **SupervisorJob** - One failure doesn't cancel others
|
||||
- **ExceptionHandler** - Log exceptions, don't crash
|
||||
- **Scoped lifecycle** - Cancel all on destroy
|
||||
|
||||
### Pattern: Retry with Backoff for Relay Connections
|
||||
|
||||
```kotlin
|
||||
fun connectWithRetry(relay: Relay): Flow<ConnectionStatus> = flow {
|
||||
var attempt = 0
|
||||
val maxRetries = 5
|
||||
val baseDelay = 1000L
|
||||
|
||||
while (attempt < maxRetries) {
|
||||
try {
|
||||
emit(ConnectionStatus.Connecting)
|
||||
relay.connect()
|
||||
emit(ConnectionStatus.Connected)
|
||||
return@flow
|
||||
} catch (e: Exception) {
|
||||
attempt++
|
||||
emit(ConnectionStatus.Error(e, attempt))
|
||||
|
||||
if (attempt < maxRetries) {
|
||||
val delay = baseDelay * (1L shl attempt) // Exponential backoff
|
||||
delay(delay)
|
||||
}
|
||||
}
|
||||
}
|
||||
emit(ConnectionStatus.Failed)
|
||||
}
|
||||
```
|
||||
|
||||
## Subscription Lifecycle
|
||||
|
||||
### Pattern: Auto-Cleanup Subscription
|
||||
|
||||
```kotlin
|
||||
@Composable
|
||||
fun ObserveRelayEvents(
|
||||
filters: List<Filter>,
|
||||
onEvent: (Event) -> Unit
|
||||
) {
|
||||
val scope = rememberCoroutineScope()
|
||||
|
||||
DisposableEffect(filters) {
|
||||
val job = scope.launch {
|
||||
relayClient.reqAsFlow(filters).collect { events ->
|
||||
events.forEach { onEvent(it) }
|
||||
}
|
||||
}
|
||||
|
||||
onDispose {
|
||||
job.cancel() // Cancels flow, triggers awaitClose
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Lifecycle:**
|
||||
1. Composable enters → subscribe
|
||||
2. filters change → cancel + re-subscribe
|
||||
3. Composable leaves → cancel + cleanup
|
||||
|
||||
### Pattern: Multiple Concurrent Subscriptions
|
||||
|
||||
```kotlin
|
||||
fun observeMultipleFeeds(
|
||||
account: Account
|
||||
): Flow<Event> = channelFlow {
|
||||
supervisorScope {
|
||||
// Home feed
|
||||
launch {
|
||||
client.reqAsFlow(filters = homeFeedFilters)
|
||||
.collect { events -> events.forEach { send(it) } }
|
||||
}
|
||||
|
||||
// Notifications
|
||||
launch {
|
||||
client.reqAsFlow(filters = notificationFilters)
|
||||
.collect { events -> events.forEach { send(it) } }
|
||||
}
|
||||
|
||||
// DMs
|
||||
launch {
|
||||
client.reqAsFlow(filters = dmFilters)
|
||||
.collect { events -> events.forEach { send(it) } }
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Benefits:**
|
||||
- All subscriptions run concurrently
|
||||
- One failure doesn't affect others (supervisorScope)
|
||||
- Single output channel for all events
|
||||
|
||||
## Performance Optimization
|
||||
|
||||
### Pattern: Shared Upstream for Multiple Collectors
|
||||
|
||||
```kotlin
|
||||
class RelayViewModel(private val client: INostrClient) : ViewModel() {
|
||||
val events: SharedFlow<Event> = client
|
||||
.reqAsFlow(relay, filters)
|
||||
.flatMapConcat { it.asFlow() }
|
||||
.shareIn(
|
||||
scope = viewModelScope,
|
||||
started = SharingStarted.WhileSubscribed(5000),
|
||||
replay = 0
|
||||
)
|
||||
}
|
||||
|
||||
// Multiple collectors share single relay subscription
|
||||
events.collect { /* UI 1 */ }
|
||||
events.collect { /* UI 2 */ }
|
||||
```
|
||||
|
||||
### Pattern: Event Deduplication Cache
|
||||
|
||||
```kotlin
|
||||
class EventCache {
|
||||
private val seen = mutableSetOf<HexKey>()
|
||||
|
||||
fun filterNew(events: List<Event>): List<Event> =
|
||||
events.filter { event ->
|
||||
if (event.id in seen) {
|
||||
false
|
||||
} else {
|
||||
seen.add(event.id)
|
||||
true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val deduplicatedEvents = relayEvents
|
||||
.map { events -> cache.filterNew(events) }
|
||||
.filter { it.isNotEmpty() }
|
||||
```
|
||||
|
||||
## Testing Relay Flows
|
||||
|
||||
### Pattern: Test with Fake Relay
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `subscription receives events`() = runTest {
|
||||
val fakeRelay = FakeRelay()
|
||||
val client = NostrClient(fakeRelay)
|
||||
|
||||
val events = mutableListOf<Event>()
|
||||
val job = launch {
|
||||
client.reqAsFlow(relay, filters).collect { list ->
|
||||
events.addAll(list)
|
||||
}
|
||||
}
|
||||
|
||||
// Simulate relay responses
|
||||
fakeRelay.sendEvent(testEvent1)
|
||||
advanceTimeBy(100)
|
||||
fakeRelay.sendEvent(testEvent2)
|
||||
advanceTimeBy(100)
|
||||
|
||||
assertEquals(2, events.size)
|
||||
job.cancel()
|
||||
}
|
||||
```
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
### ❌ Forgetting awaitClose
|
||||
|
||||
```kotlin
|
||||
// BAD: Subscription never cleaned up
|
||||
callbackFlow {
|
||||
relay.subscribe(listener)
|
||||
// Missing awaitClose!
|
||||
}
|
||||
```
|
||||
|
||||
```kotlin
|
||||
// GOOD: Proper cleanup
|
||||
callbackFlow {
|
||||
relay.subscribe(listener)
|
||||
awaitClose {
|
||||
relay.unsubscribe(listener)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### ❌ Using GlobalScope
|
||||
|
||||
```kotlin
|
||||
// BAD: Unstructured, leaks
|
||||
GlobalScope.launch {
|
||||
relay.connect()
|
||||
}
|
||||
```
|
||||
|
||||
```kotlin
|
||||
// GOOD: Scoped to lifecycle
|
||||
viewModelScope.launch {
|
||||
relay.connect()
|
||||
}
|
||||
```
|
||||
|
||||
### ❌ Blocking in Flow Operators
|
||||
|
||||
```kotlin
|
||||
// BAD: Blocks collector
|
||||
flow.map { event ->
|
||||
Thread.sleep(1000) // Blocks!
|
||||
process(event)
|
||||
}
|
||||
```
|
||||
|
||||
```kotlin
|
||||
// GOOD: Use flowOn to offload
|
||||
flow
|
||||
.map { event ->
|
||||
delay(1000) // Suspends, doesn't block
|
||||
process(event)
|
||||
}
|
||||
.flowOn(Dispatchers.Default)
|
||||
```
|
||||
@@ -0,0 +1,493 @@
|
||||
# Testing Coroutines
|
||||
|
||||
Comprehensive guide for testing async code with runTest, Turbine, and best practices.
|
||||
|
||||
## runTest - Standard Testing
|
||||
|
||||
### Basic Pattern
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `test suspend function`() = runTest {
|
||||
val result = repository.fetchData()
|
||||
assertEquals(expected, result)
|
||||
}
|
||||
```
|
||||
|
||||
**What runTest does:**
|
||||
- Skips delays automatically
|
||||
- Provides TestScope
|
||||
- Advances virtual time
|
||||
- Waits for all coroutines to complete
|
||||
|
||||
### Testing StateFlow
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `stateflow updates correctly`() = runTest {
|
||||
val viewModel = MyViewModel()
|
||||
|
||||
// Initial state
|
||||
assertEquals(UiState.Loading, viewModel.state.value)
|
||||
|
||||
// Trigger action
|
||||
viewModel.loadData()
|
||||
advanceUntilIdle() // Run all pending coroutines
|
||||
|
||||
// Verify final state
|
||||
assertEquals(UiState.Success(data), viewModel.state.value)
|
||||
}
|
||||
```
|
||||
|
||||
### Testing with Time Control
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `debounce works correctly`() = runTest {
|
||||
val viewModel = SearchViewModel()
|
||||
|
||||
viewModel.search("a")
|
||||
advanceTimeBy(100) // 100ms passed
|
||||
|
||||
viewModel.search("ab")
|
||||
advanceTimeBy(100)
|
||||
|
||||
viewModel.search("abc")
|
||||
advanceTimeBy(300) // Debounce completes
|
||||
|
||||
// Only "abc" should have triggered search
|
||||
assertEquals(listOf("abc"), viewModel.searchQueries)
|
||||
}
|
||||
```
|
||||
|
||||
**Time control functions:**
|
||||
- `advanceTimeBy(millis)` - Move virtual time forward
|
||||
- `advanceUntilIdle()` - Run all pending work
|
||||
- `runCurrent()` - Run currently scheduled tasks only
|
||||
|
||||
## Turbine - Flow Testing Library
|
||||
|
||||
### Basic Collection Testing
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `flow emits expected values`() = runTest {
|
||||
repository.observeData().test {
|
||||
assertEquals(Item1, awaitItem())
|
||||
assertEquals(Item2, awaitItem())
|
||||
assertEquals(Item3, awaitItem())
|
||||
awaitComplete()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Testing Flow Transformations
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `map transforms correctly`() = runTest {
|
||||
val source = flowOf(1, 2, 3)
|
||||
|
||||
source
|
||||
.map { it * 2 }
|
||||
.test {
|
||||
assertEquals(2, awaitItem())
|
||||
assertEquals(4, awaitItem())
|
||||
assertEquals(6, awaitItem())
|
||||
awaitComplete()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Testing Relay Subscriptions
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `relay subscription receives events`() = runTest {
|
||||
val fakeClient = FakeNostrClient()
|
||||
|
||||
fakeClient.reqAsFlow(relay, filters).test {
|
||||
// Initially empty
|
||||
assertEquals(emptyList(), awaitItem())
|
||||
|
||||
// Send event
|
||||
fakeClient.sendEvent(event1)
|
||||
assertEquals(listOf(event1), awaitItem())
|
||||
|
||||
// Send another
|
||||
fakeClient.sendEvent(event2)
|
||||
assertEquals(listOf(event1, event2), awaitItem())
|
||||
|
||||
cancelAndIgnoreRemainingEvents()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Testing Error Handling
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `catch handles errors gracefully`() = runTest {
|
||||
val errorFlow = flow {
|
||||
emit(1)
|
||||
throw IOException("Network error")
|
||||
}.catch { emit(-1) } // Fallback value
|
||||
|
||||
errorFlow.test {
|
||||
assertEquals(1, awaitItem())
|
||||
assertEquals(-1, awaitItem())
|
||||
awaitComplete()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Testing StateFlow with Turbine
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `stateflow emits updates`() = runTest {
|
||||
val viewModel = MyViewModel()
|
||||
|
||||
viewModel.state.test {
|
||||
// Skip initial value
|
||||
assertEquals(UiState.Loading, awaitItem())
|
||||
|
||||
// Trigger update
|
||||
viewModel.loadData()
|
||||
assertEquals(UiState.Success(data), awaitItem())
|
||||
|
||||
cancelAndIgnoreRemainingEvents()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Turbine assertions:**
|
||||
- `awaitItem()` - Get next emission or fail
|
||||
- `awaitComplete()` - Verify flow completed
|
||||
- `awaitError()` - Verify flow threw exception
|
||||
- `expectNoEvents()` - Assert no emissions in timeframe
|
||||
- `cancelAndIgnoreRemainingEvents()` - Stop test
|
||||
|
||||
## Testing Patterns for Amethyst
|
||||
|
||||
### Pattern: Test Relay Connection Flow
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `reconnects on connectivity change`() = runTest {
|
||||
val connectivityFlow = MutableStateFlow(ConnectivityStatus.Off)
|
||||
val relayPool = FakeRelayPool()
|
||||
|
||||
connectivityFlow
|
||||
.flatMapLatest { status ->
|
||||
when (status) {
|
||||
is ConnectivityStatus.Active -> relayPool.connectAll()
|
||||
else -> emptyFlow()
|
||||
}
|
||||
}
|
||||
.test {
|
||||
// Initially offline
|
||||
expectNoEvents()
|
||||
|
||||
// Go online
|
||||
connectivityFlow.value = ConnectivityStatus.Active(1L, false)
|
||||
assertTrue(relayPool.connected)
|
||||
|
||||
cancelAndIgnoreRemainingEvents()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Pattern: Test Event Deduplication
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `deduplicates events across relays`() = runTest {
|
||||
val relay1 = FakeRelay()
|
||||
val relay2 = FakeRelay()
|
||||
|
||||
merge(relay1.events, relay2.events)
|
||||
.distinctBy { it.id }
|
||||
.test {
|
||||
// Both relays send same event
|
||||
relay1.send(event1)
|
||||
relay2.send(event1)
|
||||
|
||||
// Only one emission
|
||||
assertEquals(event1, awaitItem())
|
||||
expectNoEvents()
|
||||
|
||||
cancelAndIgnoreRemainingEvents()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Pattern: Test Backpressure Handling
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `drops oldest events when buffer full`() = runTest {
|
||||
val fastProducer = flow {
|
||||
repeat(100) { emit(it) }
|
||||
}
|
||||
|
||||
fastProducer
|
||||
.buffer(capacity = 10, onBufferOverflow = BufferOverflow.DROP_OLDEST)
|
||||
.test {
|
||||
// Slow consumer
|
||||
delay(100)
|
||||
|
||||
// Should have dropped oldest, kept newest
|
||||
val items = mutableListOf<Int>()
|
||||
repeat(10) {
|
||||
items.add(awaitItem())
|
||||
}
|
||||
|
||||
// Newest items present
|
||||
assertTrue(90 in items)
|
||||
assertTrue(99 in items)
|
||||
|
||||
awaitComplete()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Pattern: Test Concurrent Subscriptions
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `multiple subscriptions run concurrently`() = runTest {
|
||||
val client = FakeNostrClient()
|
||||
|
||||
val feed1 = async { client.reqAsFlow(relay1, filters1).first() }
|
||||
val feed2 = async { client.reqAsFlow(relay2, filters2).first() }
|
||||
|
||||
client.sendTo(relay1, event1)
|
||||
client.sendTo(relay2, event2)
|
||||
|
||||
assertEquals(listOf(event1), feed1.await())
|
||||
assertEquals(listOf(event2), feed2.await())
|
||||
}
|
||||
```
|
||||
|
||||
## Fakes and Mocks
|
||||
|
||||
### Fake NostrClient
|
||||
|
||||
```kotlin
|
||||
class FakeNostrClient : INostrClient {
|
||||
private val subscriptions = mutableMapOf<String, MutableSharedFlow<Event>>()
|
||||
|
||||
override fun reqAsFlow(
|
||||
relay: NormalizedRelayUrl,
|
||||
filters: List<Filter>
|
||||
): Flow<List<Event>> = callbackFlow {
|
||||
val subId = RandomInstance.randomChars(10)
|
||||
val flow = MutableSharedFlow<Event>()
|
||||
subscriptions[subId] = flow
|
||||
|
||||
val events = mutableListOf<Event>()
|
||||
flow.collect { event ->
|
||||
events.add(event)
|
||||
send(events.toList())
|
||||
}
|
||||
|
||||
awaitClose {
|
||||
subscriptions.remove(subId)
|
||||
}
|
||||
}
|
||||
|
||||
fun sendEvent(event: Event) {
|
||||
subscriptions.values.forEach { it.tryEmit(event) }
|
||||
}
|
||||
|
||||
fun sendTo(relay: NormalizedRelayUrl, event: Event) {
|
||||
subscriptions[relay.url]?.tryEmit(event)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Fake Relay Pool
|
||||
|
||||
```kotlin
|
||||
class FakeRelayPool {
|
||||
var connected = false
|
||||
private val _events = MutableSharedFlow<Event>()
|
||||
val events: SharedFlow<Event> = _events.asSharedFlow()
|
||||
|
||||
fun connectAll(): Flow<Unit> = flow {
|
||||
connected = true
|
||||
emit(Unit)
|
||||
}
|
||||
|
||||
fun disconnect() {
|
||||
connected = false
|
||||
}
|
||||
|
||||
suspend fun sendEvent(event: Event) {
|
||||
_events.emit(event)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Testing Exception Handling
|
||||
|
||||
### Test CoroutineExceptionHandler
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `exception handler catches errors`() = runTest {
|
||||
val errors = mutableListOf<Throwable>()
|
||||
|
||||
val handler = CoroutineExceptionHandler { _, throwable ->
|
||||
errors.add(throwable)
|
||||
}
|
||||
|
||||
val scope = CoroutineScope(
|
||||
Dispatchers.Unconfined + SupervisorJob() + handler
|
||||
)
|
||||
|
||||
scope.launch {
|
||||
throw IOException("Test error")
|
||||
}
|
||||
|
||||
advanceUntilIdle()
|
||||
|
||||
assertEquals(1, errors.size)
|
||||
assertTrue(errors[0] is IOException)
|
||||
}
|
||||
```
|
||||
|
||||
### Test Retry Logic
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `retries failed connections`() = runTest {
|
||||
var attempts = 0
|
||||
val maxRetries = 3
|
||||
|
||||
flow {
|
||||
attempts++
|
||||
if (attempts < maxRetries) {
|
||||
throw IOException("Connection failed")
|
||||
}
|
||||
emit("Success")
|
||||
}
|
||||
.retry(maxRetries)
|
||||
.test {
|
||||
assertEquals("Success", awaitItem())
|
||||
awaitComplete()
|
||||
assertEquals(3, attempts)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Common Testing Patterns
|
||||
|
||||
### Pattern: Verify No Emissions After Cancellation
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `no emissions after cancellation`() = runTest {
|
||||
val flow = flow {
|
||||
emit(1)
|
||||
delay(1000)
|
||||
emit(2) // Should not emit
|
||||
}
|
||||
|
||||
flow.test {
|
||||
assertEquals(1, awaitItem())
|
||||
cancel()
|
||||
|
||||
// Verify no more emissions
|
||||
expectNoEvents()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Pattern: Test Time-Based Operations
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `periodic emission works`() = runTest {
|
||||
flow {
|
||||
repeat(3) {
|
||||
emit(it)
|
||||
delay(1000)
|
||||
}
|
||||
}.test {
|
||||
assertEquals(0, awaitItem())
|
||||
|
||||
advanceTimeBy(1000)
|
||||
assertEquals(1, awaitItem())
|
||||
|
||||
advanceTimeBy(1000)
|
||||
assertEquals(2, awaitItem())
|
||||
|
||||
awaitComplete()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Pattern: Test Hot Flow Conversion
|
||||
|
||||
```kotlin
|
||||
@Test
|
||||
fun `shareIn creates hot flow`() = runTest {
|
||||
var emissions = 0
|
||||
val source = flow {
|
||||
repeat(3) {
|
||||
emissions++
|
||||
emit(it)
|
||||
}
|
||||
}
|
||||
|
||||
val shared = source.shareIn(
|
||||
scope = this,
|
||||
started = SharingStarted.Eagerly,
|
||||
replay = 1
|
||||
)
|
||||
|
||||
// First collector
|
||||
shared.take(2).collect()
|
||||
assertEquals(2, emissions) // Emitted 0, 1
|
||||
|
||||
// Second collector - shares upstream
|
||||
shared.take(1).collect()
|
||||
assertEquals(3, emissions) // Only emitted 2, not restarted
|
||||
|
||||
cancel()
|
||||
}
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
1. **Use runTest for all coroutine tests**
|
||||
- Provides virtual time
|
||||
- Automatic cleanup
|
||||
|
||||
2. **Use Turbine for Flow testing**
|
||||
- Clearer assertions
|
||||
- Better error messages
|
||||
|
||||
3. **Test both success and error paths**
|
||||
- Normal flow
|
||||
- Exception handling
|
||||
- Edge cases
|
||||
|
||||
4. **Control virtual time explicitly**
|
||||
- Don't rely on real delays
|
||||
- Use `advanceTimeBy()` and `advanceUntilIdle()`
|
||||
|
||||
5. **Create fakes, not mocks**
|
||||
- Simpler to maintain
|
||||
- More realistic behavior
|
||||
- Easier to debug
|
||||
|
||||
6. **Test cancellation behavior**
|
||||
- Verify cleanup happens
|
||||
- Check no emissions after cancel
|
||||
|
||||
7. **Test concurrent operations**
|
||||
- Use `async` to spawn concurrent work
|
||||
- Verify independence with SupervisorJob
|
||||
Reference in New Issue
Block a user