Kotlin Flow

Что такое Flow и чем холодный Flow отличается от горячего?

val numbers = flow {  println("producer started")  emit(1)  emit(2)}…

Чем Flow отличается от Sequence и почему внутри билдера sequence { } нельзя вызвать delay?

// Sequence: synchronous, the consumer pulls each valueval fib = sequence {  var a = 0  var b = 1  while (true) {    yield(a)                 // pauses until the next value is pulled…

Какие операторы Flow терминальные, почему цепочка, которая заканчивается на onEach { }, ничего не делает, и что возвращает launchIn(scope)?

val prices = flowOf(1, 2, 3)  .onEach { println("got $it") }  .map { it * 10 }                        // nothing runs yet: no terminal operator suspend fun demo(scope: CoroutineScope) {  println(prices.first())                 // got 1, 10: 2 and 3 are never produced…

Чем StateFlow отличается от SharedFlow и когда использовать каждый?

class CartVM : ViewModel() {  private val _state = MutableStateFlow(CartState())  val state: StateFlow<CartState> = _state.asStateFlow()   private val _messages = MutableSharedFlow<String>(extraBufferCapacity = 1)  val messages: SharedFlow<String> = _messages.asSharedFlow()…

Найдите баг: после нажатия «Добавить» товар попадает в корзину, но список на экране не обновляется.

class CartViewModel : ViewModel() {  private val items = mutableListOf<Item>()  private val _cart = MutableStateFlow<List<Item>>(items)  val cart: StateFlow<List<Item>> = _cart.asStateFlow()   fun add(item: Item) {    items.add(item)    _cart.value = items  }}

Найдите баг: экран показывает имя пользователя, но счётчик непрочитанных, который собирается строкой ниже, так и не появляется.

class InboxFragment : Fragment(R.layout.fragment_inbox) {  private val vm: InboxViewModel by viewModels()   override fun onViewCreated(view: View, savedInstanceState: Bundle?) {    val binding = FragmentInboxBinding.bind(view)    viewLifecycleOwner.lifecycleScope.launch {      viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {        vm.userName.collect { binding.name.text = it }          // StateFlow<String>        vm.unreadCount.collect { binding.badge.text = "$it" }   // StateFlow<Int>      }    }  }}

Что такое Channel, чем он отличается от Flow и что меняют для send ёмкости RENDEZVOUS, BUFFERED, CONFLATED и UNLIMITED?

fun main() = runBlocking {  // A work queue: each job goes to exactly one worker  val jobs = Channel<Int>(Channel.BUFFERED)  repeat(2) { worker ->    launch {      for (job in jobs) {                 // ends when the channel is closed and drained…

На каком потоке выполняется каждый пронумерованный шаг этого flow и как работает flowOn?

// launched from viewModelScope (Dispatchers.Main)repo.rows()                          // flow { emit(db.readAll()) }  .map { it.toUi() }                 // (1)  .flowOn(Dispatchers.IO)  .filter { it.isNotEmpty() }        // (2)  .flowOn(Dispatchers.Default)  .collect { render(it) }            // (3)

Найдите баг: этот flow загружает две страницы параллельно и падает с IllegalStateException: Flow invariant is violated.

fun feed(api: FeedApi): Flow<List<Post>> = flow {  coroutineScope {    launch { emit(api.page(1)) }   // each page in its own coroutine    launch { emit(api.page(2)) }  }}

Что выведет этот код с zip, а что с combine? Когда вместо них нужен merge?

val numbers = flow { for (n in 1..3) { delay(100); emit(n) } }        // at 100, 200, 300 msval letters = flow { delay(150); emit("A"); delay(100); emit("B") }   // at 150, 250 ms fun main() = runBlocking {  numbers.zip(letters) { n, l -> "$n$l" }.collect { print("$it ") }  println()  numbers.combine(letters) { n, l -> "$n$l" }.collect { print("$it ") }}

flatMapConcat, flatMapMerge или flatMapLatest: что использовать для поиска по мере ввода и что ещё нужно в этом пайплайне?

@OptIn(FlowPreview::class, ExperimentalCoroutinesApi::class)class SearchVM(private val repo: SearchRepo) : ViewModel() {  private val query = MutableStateFlow("")  fun onQueryChange(text: String) { query.value = text }   val results: StateFlow<SearchState> = query…

Чем отличаются debounce, sample и throttle-first и что из них подходит для строки поиска, потока с датчика и кнопки «Оплатить»?

// kotlinx.coroutines has no throttleFirst: a few lines with flow { }fun <T> Flow<T>.throttleFirst(window: Duration): Flow<T> = flow {  var opened: TimeMark? = null              // per collector: state lives inside flow { }  collect { value ->    val mark = opened    if (mark == null || mark.elapsedNow() >= window) {…

Flow эмитит быстрее, чем сборщик успевает обрабатывать. Что происходит по умолчанию и как это меняют buffer, conflate и collectLatest?

val source = flow {  repeat(3) { delay(100); emit(it) }   // 100 ms to produce each} suspend fun compare() {  // default: about 900 ms, producer waits for the collector…

Почему catch не видит исключение из блока collect и где в цепочке flow должны стоять retry и catch?

class FeedVM(private val repo: FeedRepo) : ViewModel() {  val feed: StateFlow<FeedState> = repo.feedFlow()    .retryWhen { cause, attempt ->      if (cause !is IOException || attempt >= 3) return@retryWhen false      delay(500L * (1 shl attempt.toInt()))          // 0.5 s, 1 s, 2 s      true                                           // re-collect the upstream…

Как работают onStart, onCompletion и onEmpty и где сбрасывать флаг загрузки, чтобы он сбрасывался и при ошибке, и при отмене?

class SyncController(private val repo: SyncRepository, private val scope: CoroutineScope) {  val state = MutableStateFlow(SyncUiState())   fun start(): Job = repo.sync()                    // Flow<SyncProgress>, completes when done    .onStart { state.update { it.copy(isSyncing = true, error = null) } }    .onEach { p -> state.update { it.copy(percent = p.percent) } }…

Когда функция репозитория должна возвращать Flow<T>, а когда быть suspend fun, которая возвращает T?

Что делают stateIn и shareIn и что на самом деле контролирует SharingStarted.WhileSubscribed(5_000)?

class NotesVM(private val repo: NoteRepo) : ViewModel() {  // right: built once, one upstream collection for the ViewModel's life  val notes: StateFlow<List<Note>> = repo.notesFlow()    .stateIn(      scope = viewModelScope,      started = SharingStarted.WhileSubscribed(5_000),…

Где должна стартовать первая загрузка экрана: в init { } ViewModel, в LaunchedEffect или во flow, отданном через stateIn? Чего стоит каждый вариант?

class ProfileViewModel(private val repo: ProfileRepository) : ViewModel() {  private val retries = MutableStateFlow(0)   @OptIn(ExperimentalCoroutinesApi::class)  val state: StateFlow<ProfileUiState> = retries    .flatMapLatest {…

Один медленный подписчик шины событий на MutableSharedFlow тормозит остальных подписчиков и эмиттер. Почему и что меняют extraBufferCapacity и onBufferOverflow?

fun main() = runBlocking {  val bus = MutableSharedFlow<Int>()             // replay 0, extraBufferCapacity 0  launch { bus.collect { delay(300); println("slow got $it") } }  launch { bus.collect { println("fast got $it") } }  delay(50)                                      // let both subscribe  repeat(4) { bus.emit(it) }                     // returns only after slow takes 3…

Как Flow устроен под капотом (Flow, FlowCollector) и как написать собственный оператор?

// The whole contract, simplified from kotlinx.coroutines:// interface Flow<out T> { suspend fun collect(collector: FlowCollector<T>) }// fun interface FlowCollector<in T> { suspend fun emit(value: T) } // A custom operator: each value paired with the one before itfun <T : Any> Flow<T>.pairwise(): Flow<Pair<T, T>> = flow {…