fun main() {
    runBlocking {
        // producer
        val someEventBus = flow {
            var no = 0
            while (true) {
                emit(no++)
                delay(1.seconds)
            }
        }

        // consumer
        someEventBus
            .mapCatching {
                // error situation on consuming events
                if (it == 4) error("error")
                it
            }
            .onlySuccess()
            .onEach {
                println("Received $it")
            }
            .launchIn(this)
    }
}

sealed interface Result<R> {
    data class Success<R>(val value: R) : Result<R>
    data class Failure<R>(val cause: Throwable) : Result<R>
}

fun <T, R> Flow<T>.mapCatching(action: suspend (T) -> R) =
    map {
        try {
            Result.Success(action(it))
        } catch (ex: Throwable) {
            Result.Failure(ex)
        }
    }

fun <T> Flow<Result<T>>.onlySuccess() =
    filterIsInstance<Result.Success<T>>()
        .map { it.value }