Skip to main contentkotlin-coroutines
Advanced Kotlin coroutines patterns for AmethystMultiplatform. Use when working with: (1) Structured concurrency (supervisorScope, coroutineScope), (2) Advanced Flow operators (flatMapLatest, combine, merge, shareIn, stateIn), (3) Channels and callbackFlow, (4) Dispatcher management and context switching, (5) Exception handling (CoroutineExceptionHandler, SupervisorJob), (6) Testing async code (runTest, Turbine), (7) Nostr relay connection pools and subscriptions, (8) Backpressure handling in event streams. Delegates to kotlin-expert for basic StateFlow/SharedFlow patterns. Complements nostr-expert for relay communication.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/vitorpamplona/amethyst --skill kotlin-coroutines명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
이 저장소의 다른 Skills
Subscription and filter-assembly patterns for the Amethyst relay client layer in `commons/.../relayClient/`. Use when working with compose-scoped subscriptions (`ComposeSubscriptionManager`, `Subscribable`), filter assemblers (`MetadataFilterAssembler`, `ReactionsFilterAssembler`, `FeedMetadataCoordinator`), preloaders (`MetadataPreloader`, `MetadataRateLimiter`), EOSE managers, or any feature that needs to talk to relays lifecycle-aware from a composable. Complements `nostr-expert` (protocol filter syntax) and `kotlin-coroutines` (callbackFlow patterns).
The NIP-85 trusted-assertions model in Quartz (`nip85TrustedAssertions/`) — kind 10040 trust-provider lists, kind 30382 contact cards / user assertions, 30383 event assertions, 30384 addressable assertions, 30385 external-id assertions. Use when building or parsing these events, working with the typed tags (RankTag, HopsTag, FollowerCountTag, ServiceProviderTag/ServiceType, …), wiring a consumer that resolves a 10040 provider entry to the 30382s it signs, ranking on assertion values, or touching the GrapeRank publisher, contact-card nicknames, or the trust projection of an external store.
The NIP-50 indexing surface of Quartz — the `SearchableEvent` interface, which event kinds are searchable, exactly what text each kind's `indexableContent()` contributes, how the SQLite/filesystem stores consume it, and the NIP-50 `SearchQuery` extension grammar plus `SearchRelayListEvent` (kind 10007). Use when making a kind searchable, changing what a kind indexes, diffing the searchable set at a Quartz version bump (external search engines mirror this table), debugging why an event is or isn't found by search, or working with search extensions (`include:spam`, `domain:`, …).
| name | kotlin-coroutines |
| description | Advanced Kotlin coroutines patterns for AmethystMultiplatform. Use when working with: (1) Structured concurrency (supervisorScope, coroutineScope), (2) Advanced Flow operators (flatMapLatest, combine, merge, shareIn, stateIn), (3) Channels and callbackFlow, (4) Dispatcher management and context switching, (5) Exception handling (CoroutineExceptionHandler, SupervisorJob), (6) Testing async code (runTest, Turbine), (7) Nostr relay connection pools and subscriptions, (8) Backpressure handling in event streams. Delegates to kotlin-expert for basic StateFlow/SharedFlow patterns. Complements nostr-expert for relay communication. |
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
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) {
listOf(event) + currentEvents
} else {
currentEvents + event
}
eventIds.add(event.id)
trySend(currentEvents)
}
}
override fun onEose(...) {
hasBeenLive = true
}
}
openReqSubscription(subId, mapOf(relay to filters), listener)
awaitClose { close(subId) }
}
- Deduplication with Set
- EOSE handling (append → prepend strategy)
- trySend (non-blocking from callback)
- 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)
}
}
}
}
- One relay failure doesn't cancel others
- All cancelled together when scope cancelled
- Proper cleanup guaranteed
Pattern: Exception Handling for Services
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()
}
}
- SupervisorJob: children fail independently
- ExceptionHandler: log instead of crash
- Scoped lifecycle: cancel all on destroy
Pattern: Network Connectivity as Flow
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)
activeNetwork?.let { trySend(ConnectivityStatus.Active(...)) }
awaitClose {
connectivityManager.unregisterCallback(networkCallback)
}
}
.distinctUntilChanged()
.debounce(200)
.flowOn(Dispatchers.IO)
- Emit initial state immediately
- Register callback in flow body
- Cleanup in awaitClose
- 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 }
- 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
- Multi-relay subscription management
- Connection lifecycle and reconnection
- Event deduplication strategies
- Backpressure for high-frequency events
- EOSE handling patterns
Testing
@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()
}
}
runTest - Virtual time, auto cleanup- Turbine
.test {} - Flow assertions
advanceTimeBy() - Control time- Fake implementations over mocks
Common Scenarios
Scenario: Implement New Relay Feature
- callbackFlow for subscription
- Deduplication (Set of event IDs)
- awaitClose for cleanup
- 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
- callbackFlow for connectivity
- flatMapLatest to reconnect
- debounce to stabilize
- 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
- Use shareIn for expensive upstream
- Configure SharingStarted strategy
- Set replay buffer size
- 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
viewModelScope.launch { }
callbackFlow {
registerCallback()
}
callbackFlow {
registerCallback()
awaitClose { unregisterCallback() }
}
flow.map { Thread.sleep(1000); process(it) }
flow.map { delay(1000); process(it) }.flowOn(Dispatchers.Default)
fastProducer.collect { slowConsumer(it) }
fastProducer
.buffer(64, BufferOverflow.DROP_OLDEST)
.collect { slowConsumer(it) }
Delegation
- Basic StateFlow/SharedFlow patterns
- viewModelScope.launch usage
- Simple MutableStateFlow → asStateFlow()
- Nostr protocol details (NIPs, event structure)
- Event creation and signing
- Cryptographic operations
- 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)
