Спасибо за мнение. Пока поверю на слово. Со временем, вероятно, проникнусь идеологией

так было и с Rx

Кушал через немогу, а потом понравилось

А где вы единственное сообщение увидели? Оператор принимает три потока данных на вход и публикует целиком один из них на выход.

Он сказал, что ждем единственного сообщения и все закрываем.

select-ом под капотом это почти наверняка и делается. Создаёте сразу флоу-результат, дожидаетесь селектом первого сообщения, а дальше публикуете в ваш флоу первое сообщение, которое дождались, и все остальные.

Я не у машины, попробуйте сами поупражняться, нет -- попробую помочь позже с реализацией.

Если бы на выходе был бы Flow, то это другая ситуация, можно делать merge + debounce

Так вы на marble-диаграмму посмотрите :) Ждём-то первого сообщения, да, а потом публикуем весь флоу, откуда это сообщение пришло, забив на остальные входы.

На выходе Flow, это выплывает из диаграммы Rx

С одним элементом?

Не обязательно

А, ну тогда другой вопрос. Тогда к чему было про закрытие?

Закрыть нужно те flow, которые не выиграли

ведь значения от них уже не актуальны

Спасибо, я буду пробовать. Но пока опыта маловато

Не фига не понял, ну да ладно. Новые операторы добавляются легко. Можно сделать оператор на том же самом селекте

Будете искать закрытие -- флоу отменять можно, вам, скорее всего, именно отмена и нужна. Но сначала хотя бы без отмены сделать надо.

Грубо говоря, хочется нечто такого

       fun testFlow(name: String, timeout: Long): Flow<Int> = flow {
            for (i in 1..3) {
                delay(timeout)
                println("Emitting $i from ${name}")
                emit(i)
            }
        }

        Flow.amb(
                testFlow("First", 60),
                testFlow("Second", 50),
                testFlow("Third", 70))
               .collect { println("Received ${it}") }

чтобы вывело

Emitting 1 from Second
Received 1
Emitting 2 from Second
Received 2
Emitting 3 from Second
Received 3

       fun testFlow(name: String, timeout: Long): Flow<Int> = flow {
            for (i in 1..3) {
                delay(timeout)
                println("Emitting $i from ${name}")
                emit(i)
            }
        }

        Flow.amb(
                testFlow("First", 60),
                testFlow("Second", 50),
                testFlow("Third", 70))
               .collect { println("Received ${it}") }

Emitting 1 from Second
Received 1
Emitting 2 from Second
Received 2
Emitting 3 from Second
Received 3

Что-то в этом роде: https://pl.kotl.in/esjdXHid9, но я не тестил, и наверное там можно без канала обойтись.

Наверное, ссылка сломалась