Agent Skills: Kotlin Coroutines - Advanced Async Patterns

Kotlin Coroutines - Advanced Async Patterns

Expert guidance for complex async operations in Amethyst: relay pools, event streams, structured concurrency, and testing.

Mental Model

Async Architecture in Amethyst:

Relay Pool (supervisorScope)
    ├── Relay 1 (launch) → callbackFlow → Events
    ├── Relay 2 (launch) → callbackFlow → Events
    └── Relay 3 (launch) → callbackFlow → Events
            ↓
    merge() → distinctBy(id) → shareIn
            ↓
    Multiple Collectors (ViewModels, Services)

Key principles:

• supervisorScope - Children fail independently
• callbackFlow - Bridge callbacks to Flow
• shareIn/stateIn - Hot flows from cold
• Backpressure - buffer(), conflate(), DROP_OLDEST

When to Use This Skill

Use for advanced async patterns:

• Multi-relay subscriptions with supervisorScope
• Complex Flow operators (flatMapLatest, combine, merge)
• callbackFlow for Android callbacks (connectivity, location)
• Backpressure handling in high-frequency streams
• Exception handling with CoroutineExceptionHandler
• Testing coroutines with runTest and Turbine

Delegate to kotlin-expert for:

• Basic StateFlow/SharedFlow patterns
• Simple viewModelScope.launch
• MutableStateFlow → asStateFlow()

Core Patterns

Pattern: callbackFlow for Relay Subscriptions

// Real pattern from NostrClientStaticReqAsStateFlow.kt
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, ...) {
            if (event.id !in eventIds) {
                currentEvents = if (hasBeenLive) {
                    // After EOSE: prepend
                    listOf(event) + currentEvents
                } else {
                    // Before EOSE: append
                    currentEvents + event
                }
                eventIds.add(event.id)
                trySend(currentEvents)
            }
        }

        override fun onEose(...) {
            hasBeenLive = true
        }
    }

    openReqSubscription(subId, mapOf(relay to filters), listener)

    awaitClose { close(subId) }
}

Key techniques:

1. Deduplication with Set
2. EOSE handling (append → prepend strategy)
3. trySend (non-blocking from callback)
4. awaitClose for cleanup

Pattern: Structured Concurrency for Relays

suspend fun connectToRelays(relays: List<Relay>) = supervisorScope {
    relays.forEach { relay ->
        launch {
            try {
                relay.connect()
                relay.subscribe(filters).collect { event ->
                    eventChannel.send(event)
                }
            } catch (e: IOException) {
                Log.e("Relay", "Connection failed: ${relay.url}", e)
                // Other relays continue
            }
        }
    }
}

Why supervisorScope:

• One relay failure doesn't cancel others
• All cancelled together when scope cancelled
• Proper cleanup guaranteed

Pattern: Exception Handling for Services

// Real pattern from PushNotificationReceiverService.kt
class MyService : Service() {
    val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
        Log.e("Service", "Caught: ${throwable.message}", throwable)
    }

    private val scope = CoroutineScope(
        Dispatchers.IO + SupervisorJob() + exceptionHandler
    )

    override fun onDestroy() {
        scope.cancel()
        super.onDestroy()
    }
}

Pattern benefits:

• SupervisorJob: children fail independently
• ExceptionHandler: log instead of crash
• Scoped lifecycle: cancel all on destroy

Pattern: Network Connectivity as Flow

// Real pattern from ConnectivityFlow.kt
val status = callbackFlow {
    val networkCallback = object : NetworkCallback() {
        override fun onAvailable(network: Network) {
            trySend(ConnectivityStatus.Active(...))
        }
        override fun onLost(network: Network) {
            trySend(ConnectivityStatus.Off)
        }
    }

    connectivityManager.registerCallback(networkCallback)

    // Initial state
    activeNetwork?.let { trySend(ConnectivityStatus.Active(...)) }

    awaitClose {
        connectivityManager.unregisterCallback(networkCallback)
    }
}
    .distinctUntilChanged()
    .debounce(200)  // Stabilize flapping
    .flowOn(Dispatchers.IO)

Key patterns:

1. Emit initial state immediately
2. Register callback in flow body
3. Cleanup in awaitClose
4. Stabilize with debounce + distinctUntilChanged

Pattern: Merge Events from Multiple Relays

fun observeFromRelays(
    relays: List<NormalizedRelayUrl>,
    filters: List<Filter>
): Flow<Event> =
    relays.map { relay ->
        client.reqAsFlow(relay, filters)
            .flatMapConcat { it.asFlow() }
    }.merge()
    .distinctBy { it.id }

Flow:

• Each relay: Flow<List<Event>>
• flatMapConcat: flatten to Flow<Event>
• merge(): combine all relays
• distinctBy: deduplicate across relays

Advanced Operators

For comprehensive coverage of Flow operators:

• flatMapLatest, combine, zip, merge → See advanced-flow-operators.md
• shareIn, stateIn → Conversion to hot flows
• buffer, conflate → Backpressure strategies
• debounce, sample → Rate limiting

Quick Reference

| Operator | Use Case | Example | |----------|----------|---------| | flatMapLatest | Cancel previous, switch to new | Search (cancel old query) | | combine | Latest from ALL flows | combine(account, settings, connectivity) | | merge | Single stream from multiple | merge(relay1, relay2, relay3) | | shareIn | Multiple collectors, single upstream | Share expensive computation | | stateIn | StateFlow from Flow | ViewModel state | | buffer(DROP_OLDEST) | High-frequency streams | Real-time event feed | | conflate | Latest only | UI updates | | debounce | Wait for quiet period | Search input |

Nostr Relay Patterns

For complete relay-specific patterns: → See relay-patterns.md

Covers:

• Multi-relay subscription management
• Connection lifecycle and reconnection
• Event deduplication strategies
• Backpressure for high-frequency events
• EOSE handling patterns

Testing

For comprehensive testing patterns: → See testing-coroutines.md

Quick testing pattern:

@Test
fun `relay subscription receives events`() = runTest {
    val client = FakeNostrClient()

    client.reqAsFlow(relay, filters).test {
        assertEquals(emptyList(), awaitItem())

        client.sendEvent(event1)
        assertEquals(listOf(event1), awaitItem())

        cancelAndIgnoreRemainingEvents()
    }
}

Testing tools:

• runTest - Virtual time, auto cleanup
• Turbine .test {} - Flow assertions
• advanceTimeBy() - Control time
• Fake implementations over mocks

Common Scenarios

Scenario: Implement New Relay Feature

Steps:

1. callbackFlow for subscription
2. Deduplication (Set of event IDs)
3. awaitClose for cleanup
4. Test with FakeNostrClient

Example: Add subscription for specific event kind

fun observeKind(kind: Int): Flow<Event> = callbackFlow {
    val listener = object : IRequestListener {
        override fun onEvent(event: Event, ...) {
            if (event.kind == kind) {
                trySend(event)
            }
        }
    }
    client.subscribe(listener)
    awaitClose { client.unsubscribe(listener) }
}

Scenario: Handle Network Connectivity Changes

Steps:

1. callbackFlow for connectivity
2. flatMapLatest to reconnect
3. debounce to stabilize
4. Exception handling for failures

Example: Reconnect relays on connectivity

connectivityFlow
    .flatMapLatest { status ->
        when (status) {
            Active -> relayPool.observeEvents()
            else -> emptyFlow()
        }
    }
    .catch { e -> Log.e("Error", e) }
    .collect { event -> handleEvent(event) }

Scenario: Optimize Multi-Collector Performance

Steps:

1. Use shareIn for expensive upstream
2. Configure SharingStarted strategy
3. Set replay buffer size
4. Test with multiple collectors

Example: Share relay subscription

val events: SharedFlow<Event> = client
    .reqAsFlow(relay, filters)
    .flatMapConcat { it.asFlow() }
    .shareIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5000),
        replay = 0
    )

Anti-Patterns

❌ Using GlobalScope

GlobalScope.launch { /* Leaks, no structured concurrency */ }

✅ Use scoped coroutines

viewModelScope.launch { /* Cancelled with ViewModel */ }

❌ Forgetting awaitClose

callbackFlow {
    registerCallback()
    // Missing cleanup!
}

✅ Always cleanup

callbackFlow {
    registerCallback()
    awaitClose { unregisterCallback() }
}

❌ Blocking in Flow

flow.map { Thread.sleep(1000); process(it) }

✅ Suspend, don't block

flow.map { delay(1000); process(it) }.flowOn(Dispatchers.Default)

❌ Ignoring backpressure

fastProducer.collect { slowConsumer(it) }  // Blocks producer!

✅ Handle backpressure

fastProducer
    .buffer(64, BufferOverflow.DROP_OLDEST)
    .collect { slowConsumer(it) }

Delegation

Use kotlin-expert for:

• Basic StateFlow/SharedFlow patterns
• viewModelScope.launch usage
• Simple MutableStateFlow → asStateFlow()

Use nostr-expert for:

• Nostr protocol details (NIPs, event structure)
• Event creation and signing
• Cryptographic operations

This skill provides:

• Advanced async patterns
• Structured concurrency
• Complex Flow operators
• Testing strategies
• Relay-specific async patterns

Resources

• references/advanced-flow-operators.md - All Flow operators with examples
• references/relay-patterns.md - Nostr relay async patterns from codebase
• references/testing-coroutines.md - Complete testing guide

Quick Decision Tree

Need async operation?
    ├─ Simple ViewModel state update → kotlin-expert (StateFlow)
    ├─ Android callback → This skill (callbackFlow)
    ├─ Multiple concurrent operations → This skill (supervisorScope)
    ├─ Complex Flow transformation → This skill (references/advanced-flow-operators.md)
    ├─ Relay subscription → This skill (references/relay-patterns.md)
    └─ Testing async code → This skill (references/testing-coroutines.md)