Learn how to code libraries that your iOS teammates will not frown upon using them. This is the final chapter: Flow!

An Android controlling a Flow of apples — DALLE-2

I know you have been waiting long for the Great Season Finale of this series: Flow! Let us see if we can apply what we learned so far.

This article is part of a series, see the other articles here

I will use the most classical example: a flow that every second emits a value. In this case, I am using kotlinx-datetime, the official multiplatform library for time.

// KOTLIN API CODE
class Clock {
    fun ticker(): Flow<String> = flow {
        while (currentCoroutineContext().isActive) {
            val now: Instant = Clock.System.now()
            val thisTime: LocalTime = now.toLocalDateTime(TimeZone.currentSystemDefault()).time
            emit(thisTime.toString())
            delay(1_000)
        }
    }
}

// EXPORTED OBJ-C HEADER
__attribute__((objc_subclassing_restricted))
__attribute__((swift_name("Watch")))
@interface SharedWatch : SharedBase
- (instancetype)init __attribute__((swift_name("init()"))) __attribute__((objc_designated_initializer));
+ (instancetype)new __attribute__((availability(swift, unavailable, message="use object initializers instead")));
- (id<SharedKotlinx_coroutines_coreFlow>)ticker __attribute__((swift_name("ticker()")));
@end

__attribute__((swift_name("Kotlinx_coroutines_coreFlow")))
@protocol SharedKotlinx_coroutines_coreFlow
@required

/**
 * @note This method converts instances of CancellationException to errors.
 * Other uncaught Kotlin exceptions are fatal.
*/
- (void)collectCollector:(id<SharedKotlinx_coroutines_coreFlowCollector>)collector completionHandler:(void (^)(NSError * _Nullable))completionHandler __attribute__((swift_name("collect(collector:completionHandler:)")));
@end

__attribute__((swift_name("Kotlinx_coroutines_coreFlowCollector")))
@protocol SharedKotlinx_coroutines_coreFlowCollector
@required

/**
 * @note This method converts instances of CancellationException to errors.
 * Other uncaught Kotlin exceptions are fatal.
*/
- (void)emitValue:(id _Nullable)value completionHandler:(void (^)(NSError * _Nullable))completionHandler __attribute__((swift_name("emit(value:completionHandler:)")));
@end

// HEADER "TRANSLATED" TO SWIFT 
public class Watch : KotlinBase {
    public init()
    open func ticker() -> Kotlinx_coroutines_coreFlow
}

public protocol Kotlinx_coroutines_coreFlow {
    /**
     * @note This method converts instances of CancellationException to errors.
     * Other uncaught Kotlin exceptions are fatal.
    */
    func collect(collector: Kotlinx_coroutines_coreFlowCollector, completionHandler: @escaping (Error?) -> Void)

    /**
     * @note This method converts instances of CancellationException to errors.
     * Other uncaught Kotlin exceptions are fatal.
    */
    func collect(collector: Kotlinx_coroutines_coreFlowCollector) async throws
}

public protocol Kotlinx_coroutines_coreFlowCollector {
    /**
     * @note This method converts instances of CancellationException to errors.
     * Other uncaught Kotlin exceptions are fatal.
    */
    func emit(value: Any?, completionHandler: @escaping (Error?) -> Void)

    /**
     * @note This method converts instances of CancellationException to errors.
     * Other uncaught Kotlin exceptions are fatal.
    */
    func emit(value: Any?) async throws
}

So what happened to our beautiful KMP API? In Part VIII we saw that Kotlin Interfaces are exported as Objective-C Protocols, which do not support Generics. Consequently, the type argument String is lost. The collect function is a suspend method, and as we saw in Part VII it will be exported to Objective-C as a function with a completion handler. Additionally, thanks to Xcode bridging, it will also feature an async version for Swift.

Although the API is somewhat usable, our Swift teammate will have two major inconveniences:

• Because the Generic type information is lost, the Swift developer will have to forcibly cast (as!) every flow emission from Any? to String;
• There is no way to cancel our infinite flow.

With all those problems, how can we convince iOS developers that KMP is a good idea?

A Solution

Based on the last two chapters of this series we know that a solution will require two things:

1. We must return a class because Kotlin classes are exported as Objective-C classes, the only type that really supports generics;
2. We need to create a wrapper for Flow that works like a handle, having one method to start the flow and another to cancel it.

FlowWrapper as the name says will work as our handle class. The constructor will receive the flow to be wrapped and the coroutine scope that we want to use for the collection. Its collect method will follow the same pattern as all other solutions that I saw for flows. It will have a very “ReactiveX” signature with an onEach callback for emissions and an onComplete callback for errors and completion. In fact, most solutions that I saw name such method subscribe.

In the iosMain sourceset we create the extension Watch.wrappedTicker and we use @ObjCName(“ticker”) on it and @HiddenFromObjC on the original method to effectively replace it with the wrapped version for iOS.

// KOTLIN API CODE
class Watch {
    @OptIn(ExperimentalObjCRefinement::class)
    @HiddenFromObjC // Will not be exported to Obj-C and Swift
    fun ticker(): Flow<String> = flow {
        while (currentCoroutineContext().isActive) {
            val now: Instant = Clock.System.now()
            val thisTime: LocalTime = now.toLocalDateTime(TimeZone.currentSystemDefault()).time
            emit(thisTime.toString())
            delay(1_000)
        }
    }
}

class FlowWrapper<out T> internal constructor(private val scope: CoroutineScope,
                                              private val flow: Flow<T & Any>){
    private var job: Job? = null
    private var isCancelled = false

    /**
     *  Cancels the flow
     */
    fun cancel() {
        isCancelled = true
        job?.cancel()
    }

    /**
     * Starts the flow
     * @param onEach callback called on each emission
     * @param onCompletion callback called when flow completes. It will be provided with a non
     * nullable Throwable if it completes abnormally
     */
    fun collect(
        onEach: (T & Any) -> Unit,
        onCompletion: (Throwable?) -> Unit
    ) {
        if (isCancelled) return
        job = scope.launch {
            flow.onEach(onEach).onCompletion { cause: Throwable? -> onCompletion(cause) }.collect()
        }
    }
}

internal fun <T> Flow<T&Any>.wrap(scope: CoroutineScope = MainScope()) = FlowWrapper(scope, this)

// iosMain
@ObjCName("ticker")
fun Watch.wrappedTicker() = this.ticker().wrap()

// EXPORTED OBJ-C HEADER
__attribute__((objc_subclassing_restricted))
__attribute__((swift_name("Watch")))
@interface SharedWatch : SharedBase
- (instancetype)init __attribute__((swift_name("init()"))) __attribute__((objc_designated_initializer));
+ (instancetype)new __attribute__((availability(swift, unavailable, message="use object initializers instead")));
@end

__attribute__((objc_subclassing_restricted))
__attribute__((swift_name("FlowWrapper")))
@interface SharedFlowWrapper<__covariant T> : SharedBase
- (void)cancel __attribute__((swift_name("cancel()")));
- (void)collectOnEach:(void (^)(T))onEach onCompletion:(void (^)(SharedKotlinThrowable * _Nullable))onCompletion __attribute__((swift_name("collect(onEach:onCompletion:)")));
@end

@interface SharedWatch (Extensions)
- (SharedFlowWrapper<NSString *> *)ticker __attribute__((swift_name("ticker()")));
@end

// HEADER "TRANSLATED" TO SWIFT 
public class Watch : KotlinBase {
    public init()
}

public class FlowWrapper<T> : KotlinBase where T : AnyObject {
    open func cancel()
    open func collect(onEach: @escaping (T) -> Void, onCompletion: @escaping (KotlinThrowable?) -> Void)
}

extension Watch {
    open func ticker() -> FlowWrapper<NSString>
}

Now our Swift colleague can not only collect the flow but also be notified that it has been completed and cancel it.

// SWIFT CLIENT CODE
func test() {
    let watch = Watch()
    let handle = watch.ticker()
    handle.collect {
        value in print(value)
    } onCompletion: { throwable in
        print("Complete : \(throwable?.description() ?? "<Success>")")
    }

    DispatchQueue.main.asyncAfter(deadline: .now() + 10){
        print("CANCELLING")
        handle.cancel()
    }
}

But can we make our iOS dev colleague feel more at home?

Being more “Swifty”: Supporting Combine

In the same way, we Android developers are moving from RxJava to Flows, Swift developers are moving from RxSwift to Combine — Apple’s implementation of reactive streams.

However, Combine is a pure Swift Framework and consequently, it is not accessible from Objective-C. In that case, we learned in Part IV that in those cases we need to get our hands dirty and write a Swift extension to fill the gaps in our API. Let’s do it then!

First, we extend KotlinThrowable — the exported version of Throwable — to conform to the Error protocol so we can forward any exceptions on our flow. Second, we need to implement a Publisher that will wrap our wrapper to our flow (“We can solve any problem by introducing an extra level of indirection.”). Combine is based on the Publisher-Subscriber paradigm. This sequence diagram gives us a good idea of how those two interact:

Sequence diagram for Combine Publisher-Subscriber interaction

Next, our publisher defines the Associated Types for the Publisher protocol by setting Output to our generic type argument and Failure to KotlinThrowable. Then, we need to implement the method receive, which will have to follow the sequence depicted in the diagram above:

1. Calls Subscriber::receive(subscription:)to pass a Subscription object. We have created FlowSubscription, which conforms to the Subscription protocol, so we can relay the subscription cancellation requests to our flow;
2. Call Subscriber::receive(_ input:) to forward the elements emitted by the flow;
3. Finally, call Subscriber::receive(completion:) when the flow ends either normally (Subscribers.Completion::finished) or with failure (Subscriber.Completion::failure(Failure))

import Combine

extension KotlinThrowable: Error {}

class FlowPublisher<T: AnyObject>: Publisher {
    typealias Output = T
    typealias Failure = KotlinThrowable
    
    private let wrappedFlow: FlowWrapper<T>
    
    init(wrappedFlow: FlowWrapper<T>) {
        self.wrappedFlow = wrappedFlow
    }
    
    func receive<S>(subscriber: S) where S : Subscriber, KotlinThrowable == S.Failure, T == S.Input {
        let subscription = FlowSubscription(wrappedFlow: wrappedFlow)
        
        subscriber.receive(subscription: subscription)
        
        wrappedFlow.collect { value in
           let demand: Subscribers.Demand = subscriber.receive(value)
           // Dealing with demand is left as exercise for the reader
        } onCompletion: { throwable in
            subscriber.receive(completion: throwable == nil ? .finished : .failure(throwable!))
        }
    }
    
    class FlowSubscription: Subscription {
        
        private let wrappedFlow: FlowWrapper<T>
        
        init(wrappedFlow: FlowWrapper<T>) {
            self.wrappedFlow = wrappedFlow
        }
        
        func request(_ demand: Subscribers.Demand) {
            // Dealing with demand is left as exercise for the reader
        }
        
        //Progates the cancel 
        func cancel() {
            wrappedFlow.cancel()
        }
    }
}

func flow<T>(_ wrapper: FlowWrapper<T>) -> FlowPublisher<T> {
    return FlowPublisher(wrappedFlow: wrapper)
}

The flow above is a helper function to wrap our wrapper. Now Swift devs can consume our flow in a more familiar way:

func test() {
    let watch = Watch()
    let handle = flow(watch.ticker())
        .handleEvents(receiveSubscription: { _ in print("Subscribed") },
                      receiveCancel: { print("Cancelled") })
        .sink(receiveCompletion: { completion in switch completion {
        case .finished:
            print("Completed with success")
            break
        case let .failure(throwable):
            print("Completed with failure: \(throwable)")
            break
        }}, receiveValue: { value in print(value) })
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 10){
        print("CANCELLING")
        handle.cancel()
    }
}

Subscribers.Demand is the Combine way to deal with backpressure.

KMP-NativeCoroutines

Of course, like for coroutines, people already created libraries so you do not have to learn Swift. The libraries that I cited in Part VII, SKIE, Koru, and KMP-NativeCoroutines also support Flows. Again, let us see how we can use KMP-NativeCoroutines with our example.

Usage is very similar to coroutines. We use the @NativeCoroutineScope annotation to denote the scope that we want to use to collect the flows. Again @NativeCoroutines will hide our original method, and during the annotation processing phase by KSP it will provoke the generation of the replacement extension Watch.tickerNative().

// KOTLIN API CODE
@NativeCoroutineScope
internal val coroutineScope = MainScope()

class Watch {
    @NativeCoroutines
    fun ticker(): Flow<String> = flow {
        while (currentCoroutineContext().isActive) {
            val now: Instant = Clock.System.now()
            val thisTime: LocalTime = now.toLocalDateTime(TimeZone.currentSystemDefault()).time
            emit(thisTime.toString())
            delay(1_000)
        }
    }
}

// ios* CODE GENERATED BY KSP
@ObjCName(name = "ticker")
public fun Watch.tickerNative(): NativeFlow<String> = ticker().asNativeFlow(coroutineScope)

// EXPORTED OBJ-C HEADER
__attribute__((objc_subclassing_restricted))
__attribute__((swift_name("Watch")))
@interface SharedWatch : SharedBase
- (instancetype)init __attribute__((swift_name("init()"))) __attribute__((objc_designated_initializer));
+ (instancetype)new __attribute__((availability(swift, unavailable, message="use object initializers instead")));
@end

@interface SharedWatch (Extensions)
- (SharedKotlinUnit *(^(^)(SharedKotlinUnit *(^)(NSString *, SharedKotlinUnit *(^)(void), SharedKotlinUnit *), SharedKotlinUnit *(^)(NSError * _Nullable, SharedKotlinUnit *), SharedKotlinUnit *(^)(NSError *, SharedKotlinUnit *)))(void))ticker __attribute__((swift_name("ticker()")));
@end

// HEADER "TRANSLATED" TO SWIFT 
public class Watch : KotlinBase {
    public init()
}

extension Watch {
    open func ticker() -> (@escaping (String, @escaping () -> KotlinUnit, KotlinUnit) -> KotlinUnit, @escaping (Error?, KotlinUnit) -> KotlinUnit, @escaping (Error, KotlinUnit) -> KotlinUnit) -> () -> KotlinUnit
}

In a similar way to our solution, we need to use createPublisher(for:) to wrap the flow into a publisher.

import KMPNativeCoroutinesCombine

func test() {
    let watch = Watch()
    let publisher = createPublisher(for: watch.ticker())
    let handle = publisher
        .handleEvents(receiveSubscription: { _ in print("Subscribed") },
                      receiveCancel: { print("Cancelled") })
        .sink(receiveCompletion: { completion in switch completion {
        case .finished:
            print("Completed with success")
            break
        case let .failure(throwable):
            print("Completed with failure: \(throwable)")
            break
        }}, receiveValue: { value in print(value) })
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 10){
        print("CANCELLING")
        handle.cancel()
    }
}

And this is the end! C’est la fin!

Now you can appreciate the entire series at once.

Ci vediamo a Torino! See y’all at London!

Droidcon Italy 2023

Droidcon London 2023

As usual, every June I attend Droidcon San Francisco. As I got my seat for one of the first talks, I started hearing some Portuguese. I learned it was a group of Brazilian compatriots who work in Europe and they came to Droidcon SF as speakers. After we exchanged our experiences of working in Europe and Silicon Valley, they really convinced me to be a speaker as well.

Brazilians at Droidcon San Francisco 2023 — Myself, Jeremy Lee (not a Brazilian but we adopted him 🙂), Robson S Lima, Daniel Horowitz , Iury Souza , and Zhenlei Ji who is taking this picture.

This article was previously published on proandroiddev.com