Coroutines Flow Pattern

1import kotlinx.coroutines.*
2import kotlinx.coroutines.flow.*
3 
4data class Event(val type: String, val payload: String)
5 
6class EventStream {
7    private val _events = MutableSharedFlow<Event>()
8    val events: SharedFlow<Event> = _events.asSharedFlow()
9 
10    suspend fun emit(event: Event) {
11        _events.emit(event)
12    }
13}
14 
15fun fetchUsers(): Flow<User> = flow {
16    listOf("Alice", "Bob", "Charlie").forEach { name ->
17        delay(100)
18        emit(User(name.hashCode(), name, "$name@example.com"))
19    }
20}
21 
22suspend fun processEvents() = coroutineScope {
23    val stream = EventStream()
24    
25    launch {
26        stream.events
27            .filter { it.type == "user" }
28            .map { it.payload.uppercase() }
29            .collect { println("Received: $it") }
30    }
31 
32    launch {
33        fetchUsers()
34            .onEach { stream.emit(Event("user", it.name)) }
35            .catch { e -> println("Error: ${e.message}") }
36            .collect()
37    }
38}