Listen to Kotlin coroutine flow from iOS

Question

I have setup a Kotlin Multiplatform project and attached a SQLDelight database to it. Its all setup and running correctly as i have tested it on the android side using the following:

commonMain:

    val backgroundColorFlow: Flow<Color> =
            dbQuery.getColorWithId(BGColor.id)
                    .asFlow()
                    .mapToOneNotNull()

which triggers fine in the Android projects MainActivity.kt using:

database.backgroundColorFlow.onEach { setBackgroundColor(it.hex) }.launchIn(lifecycleScope)

but when trying to access the same call in the iOS projects app delegate i get the following options and im unsure how to use them or convert them into my BGColor object:

database.backgroundColorFlow.collect(collector: T##Kotlinx_coroutines_coreFlowCollector, completionHandler: (KotlinUnit?, Error?) -> Void)

can anyone help me with how to use this?

Answer 1

So this was resolved by creating a flow helper:

import io.ktor.utils.io.core.Closeable
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun <T> Flow<T>.asCommonFlow(): CommonFlow<T> = CommonFlow(this)
class CommonFlow<T>(private val origin: Flow<T>) : Flow<T> by origin {
    fun watch(block: (T) -> Unit): Closeable {
        val job = Job()

        onEach {
            block(it)
        }.launchIn(CoroutineScope(Dispatchers.Main + job))

        return object : Closeable {
            override fun close() {
                job.cancel()
            }
        }
    }
}

My backgroundColorFlow var is update as follows to utilise this helper:

    val backgroundColorFlow: CommonFlow<BGColor> =
            dbQuery.getColorWithId(BGColor.id)
                    .asFlow()
                    .mapToOneNotNull()
                    .map { BGColor(it.name) }
                    .asCommonFlow()

Then my swift works as follows:

database.backgroundColorFlow.watch { color in
            guard let colorHex = color?.hex else {
                return
            }
            self.colorBehaviourSubject.onNext(colorHex)
        }

and android like so:

database.backgroundColorFlow.watch { setBackgroundColor(it.hex) }

Hope this helps anyone that comes across this. I would like to convert the CommonFlow class into an extension of Flow but don't have the know-how atm so if any could that IMHO would be a much nicer solution

Answer 2

You can do it in swift, with the mentioned collect method FlowCollector is a protocol which can be implemented to collect the data of the Flow object.

Generic example implementation could look like:

class Collector<T>: FlowCollector {

    let callback:(T) -> Void

    init(callback: @escaping (T) -> Void) {
        self.callback = callback
    }


    func emit(value: Any?, completionHandler: @escaping (KotlinUnit?, Error?) -> Void) {
        // do whatever you what with the emitted value
        callback(value as! T)

        // after you finished your work you need to call completionHandler to 
        // tell that you consumed the value and the next value can be consumed, 
        // otherwise you will not receive the next value
        //
        // i think first parameter can be always nil or KotlinUnit()
        // second parameter is for an error which occurred while consuming the value
        // passing an error object will throw a NSGenericException in kotlin code, which can be handled or your app will crash
        completionHandler(KotlinUnit(), nil) 
    }
}

The second part is calling the Flow.collect function

database.backgroundColorFlow.collect(collector: Collector<YourValueType> { yourValue in 
    // do what ever you want
}) { (unit, error) in 
    // code which is executed if the Flow object completed 
}

probably you also like to write some extension function to increase readability

Answer 3

I had to deal with that today as well so I copied the kotlin NopCollector and created own collect() extension function in Swift:

import Foundation
import shared

/**
 * Copy of kotlinx.coroutines.flow.internal.NopCollector
 */
class NopCollector<T>: Kotlinx_coroutines_coreFlowCollector { 
    func emit(value: Any?) async throws {
        // does nothing
    }
}

/**
 * Copy of kotlinx.coroutines.flow.collect extension function
 */
extension Kotlinx_coroutines_coreFlow {
    func collect() async {
        try? await collect(collector: NopCollector<Any?>())
    }
}

With this you can simply collect your flow as follows in Swift:

  Task {
        let messageStream = myService.receiveMessageStream(listener: self)
        await messageStream.collect()
  }

As you can see in receiveMessageStream is passed listener (kotlin Interface from shared module) which you can implement in View Model (ObservableObject) as classic swift Protocol. This is to receive all the events of the kotlin flow. Here is how to propagate them in shared module in kotlin:

fun receiveMessageStream(listener: IWebsocketApi.IListener): Flow<Message> {
        return flow {
            createSessionWithWebSocket(this, listener)
        }.onStart {
            listener.onWsStart()
        }.onEach {
            listener.onWsEach(it)
        }.catch {
            listener.onWsConnectionFailure(it)
        }.onCompletion {
            listener.onWsCompletion(it)
        }
    }

And that's how you Listen to Kotlin coroutine flow from iOS. EDIT

Also this is needed in your gradle.properties: kotlin.native.binary.objcExportSuspendFunctionLaunchThreadRestriction=none