Modern Android codebases are becoming increasingly reactive. With concepts and patterns such as MVI, Redux, Unidirectional Data Flow, many components of the system are being modelled as streams.

UI events can also be modelled as streams of inputs into the system.

Android’s platform and unbundled UI widgets provide listener / callback style APIs, but with RxBinding they can easily be mapped to RxJava Observable .

findViewById<Button>(R.id.button).clicks().subscribe {

// handle button clicked

}

Kotlin Flow

kotlinx.coroutines 1.3 introduced Flow , which is an important addition to the library which finally has support for cold streams. It’s (conceptually) a reactive streams implementation based on Kotlin’s suspending functions and channels API.

Binding Android UI with Flow

In this post I’m not going to discuss why you may or may not want to migrate from RxJava to Kotlin Coroutines / Flow. But let’s see how we can implement the same clicks() example above with Flow . The API should look something like this:

scope.launch {

findViewById<Button>(R.id.button)

.clicks() // this returns a Flow<Unit>

.collect {

// handle button clicked

}

}

The kotlinx.coroutines library offers many top-level builder functions for creating Flow . One such function is callbackFlow which is specifically designed for converting a multi-shot callback API into a Flow .

fun View.clicks(): Flow<Unit> = callbackFlow

val listener = View.OnClickListener {

offer(Unit)

}

setOnClickListener(listener)

awaitClose { setOnClickListener(null) }

}

The block within awaitClose is run when the consumer of the flow cancels the flow collection so this is the perfect place to remove the listener registered earlier.

offer(…) pushes a new element into the SendChannel which Flow uses internally. But the function might throw an exception if the channel is closed for send. We can create an extension function that catches any cancellation exception:

fun <E> SendChannel<E>.safeOffer(value: E) = !isClosedForSend && try {

offer(value)

} catch (e: CancellationException) {

false

}

Here’s the complete implementation:

Some UI widgets might hold a state internally such as the current value of a Slider (a recently added Material Component) which you might want to observe with a Flow . In this case it might also be useful if the Flow can emit the current value immediately when collected, so that you can bind the value to some other UI element as soon as the screen is launched without the value of the slider ever being changed by the user.

@CheckResult

@UseExperimental(ExperimentalCoroutinesApi::class)

fun Slider.valueChanges(emitImmediately: Boolean = false): Flow<Float> = callbackFlow {

checkMainThread()

val listener = Slider.OnChangeListener { _, value ->

safeOffer(value)

}

setOnChangeListener(listener)

awaitClose { setOnChangeListener(null) }

}

.startWithCurrentValue(emitImmediately) { value }

.conflate()

The optional emitImmediately parameter controls whether to emit the current value immediately on flow collection.

When emitImmediately is true we add onStart { emit(value)} on the original flow which is the equivalent of startWith(value) in RxJava. This behaviour can again be wrapped in an extension function:

fun <T> Flow<T>.startWithCurrentValue(emitImmediately: Boolean, block: () -> T?): Flow<T> {

return if (emitImmediately) onStart {

block()?.run { emit(this) }

} else this

}

As we can see it’s quite easy to implement UI event bindings for Kotlin Flow, thanks to the powerful Coroutines APIs. But there are myriad of other widgets both from the platform and the unbundled libraries (AndroidX), while new components such as MaterialDatePicker and Slider are being added to Material Components Android.

It’d be nice if we have a library of these bindings for Kotlin Flow.