Так у вас на один вызов parse() много вызовов коллбека, т.е. несколько результирующих сообщений?

Не-не, один parse - один вызов одного из коллбэков

Тогда вам Flow совершенно не нужен

Он решает проблему с повторным созданием парсера на каждое сообщение. Разве нет?

Вообще он совсем не об этом, и да, я не представляю, как он может помочь решить эту проблему.

Если вы только не хотите свалить все результаты в одну кучу 😄

Ну как-то вот так:

abstract class BaseMessageType {}
class SomeType: BaseMessageType() {}
class OtherType: BaseMessageType() {}

interface UserCallbacks {
    fun onSomeMessage(msg: SomeType)
    fun onOtherMessage(msg: OtherType)
}

class LibraryApi(val callbacks: UserCallbacks) {
    fun parse(msg: Array<Byte>) {
        if (msg.size % 2 == 1) callbacks.onSomeMessage(SomeType())
        else callbacks.onOtherMessage(OtherType())
    }
}

fun Flow<Array<Byte>>.parse(): Flow<BaseMessageType> = callbackFlow {
    val callbacks = object: UserCallbacks {
        override fun onSomeMessage(msg: SomeType) { offer(msg) }
        override fun onOtherMessage(msg: OtherType) { offer(msg) }
    }
    val parser = LibraryApi(callbacks)
    collect { bytes -> parser.parse(bytes) }
}

Видимо хочу )

abstract class BaseMessageType {}
class SomeType: BaseMessageType() {}
class OtherType: BaseMessageType() {}

interface UserCallbacks {
    fun onSomeMessage(msg: SomeType)
    fun onOtherMessage(msg: OtherType)
}

class LibraryApi(val callbacks: UserCallbacks) {
    fun parse(msg: Array<Byte>) {
        if (msg.size % 2 == 1) callbacks.onSomeMessage(SomeType())
        else callbacks.onOtherMessage(OtherType())
    }
}

fun Flow<Array<Byte>>.parse(): Flow<BaseMessageType> = callbackFlow {
    val callbacks = object: UserCallbacks {
        override fun onSomeMessage(msg: SomeType) { offer(msg) }
        override fun onOtherMessage(msg: OtherType) { offer(msg) }
    }
    val parser = LibraryApi(callbacks)
    collect { bytes -> parser.parse(bytes) }
}

Это немного отличается от изначального интерфейса suspend fun parse(bytes): Message

Ну, я запостил задачу-минимум, а мне подсказали сразу решение задачи-максимум

Тогда да, Flow - это то, что нужно. Но надо понимать, что порядок элементов в исходном и результирующем Flow могут отличаться.

Значит, @noraltavir хорошо прокачал телепатию))

ну так. Single-fire callback - это вообще редкая штука.

Да не такая уж редкая, куча таких асинхронных API с методами вида void doSomething(args, callback)

Для этого не делают интерфейс с отпиской

Если есть подписка, уже не single-fire

Ну, совсем идеально было бы всё-таки flow-версию построить через single, хотя бы для юнит-тестов каких-нибудь могло пригодиться потом. В конце концов, предметка не принуждает оперировать последовательностями вместо отдельных сообщений. Это уже ограничения имплементации диктуют

Никто не мешает делать single-fire flow

Перфекционизм мешает ) Будет зудеть, что на каждое сообщение лишний объект создаётся. Либо костыли через поле, хранящее var Deferred<Result>

Это не проблема: flowOf(foo).parse().toList()