If you’re a mid-level developer, you probably have a solid grasp of Kotlin basics, coroutines, and maybe even some experience with Flows. But if you’ve ever struggled with integrating legacy callback APIs (like those from Android’s Location Services or third-party libraries) into a coroutine-based architecture, this post is for you. We’ll explore what callbackFlow is, why it’s useful, how to use it, and I’ll provide full code examples to make it crystal clear. By the end, you’ll be confident in using it to make your Android apps more reactive and maintainable.
This post is structured to build your understanding step by step. We’ll start with the fundamentals, move into practical usage, cover real-world Android scenarios, discuss best practices and common pitfalls, and wrap up with some advanced tips. Aim for around 2000 words? Buckle up—we’re going in-depth!
Understanding the Basics: Callbacks vs. Flows
Before we jump into callbackFlow, let’s ensure we’re on the same page with the core concepts. As a mid-level dev, you know that asynchronous programming is key in Android to avoid blocking the UI thread. Traditionally, this was handled with callbacks—functions passed as parameters that get invoked when an operation completes.
What Are Callbacks?
Callbacks are essentially listener interfaces or lambdas that you register with an API. For example, in Android’s LocationManager, you might implement a LocationListener to receive location updates:
Kotlin
val locationListener = object : LocationListener {
override fun onLocationChanged(location: Location) {
// Handle new location
}
// Other overrides...
}
locationManager.requestLocationUpdates(LocationManager.GPS_PROVIDER, 0L, 0f, locationListener)
This works, but it has drawbacks: callback hell (nested callbacks), error handling scattered across methods, and it’s not composable. Plus, managing lifecycle (like removing listeners) can lead to memory leaks if not done carefully.
Enter Kotlin Flows
Flows, part of Kotlin Coroutines, represent a stream of values that can be asynchronously computed and emitted over time. They’re reactive, cancellable, and integrate seamlessly with coroutines. A simple Flow might look like this:
Kotlin
fun simpleFlow(): Flow<Int> = flow {
for (i in 1..5) {
delay(1000) // Simulate async work
emit(i)
}
}
// Collecting it
lifecycleScope.launch {
simpleFlow().collect { value ->
Log.d("Flow", "Received: $value")
}
}
Flows shine in handling sequences of data, like network responses or sensor readings. They’re cold by default (start emitting only when collected), support operators like map, filter, and debounce, and handle backpressure elegantly.
But what if your data source is callback-based? That’s where callbackFlow comes in—it’s a builder function that lets you create a Flow from a callback API.
Introducing callbackFlow: The Bridge Between Worlds
callbackFlow is a function in the kotlinx.coroutines.flow package that creates a Flow backed by a channel. It allows you to produce values inside a suspendable block using trySend (or awaitClose for cleanup), effectively turning imperative callback registrations into declarative Flows.
The basic structure:
Kotlin
fun <T> callbackFlow(block: suspend ProducerScope<T>.() -> Unit): Flow<T>
- ProducerScope<T>: An extension of CoroutineScope with methods like trySend (non-suspending send) and channel for direct access.
- Inside the block, you register your callback, and when data arrives, you trySend it to the Flow.
- Crucially, use awaitClose { /* cleanup */ } to unregister callbacks when the Flow is cancelled or completed.
Why trySend instead of emit? Because callbackFlow uses a channel under the hood, and trySend is non-suspending, which is perfect for callbacks that might be called from any thread.
Why Use callbackFlow?
- Seamless Integration: Convert old-school APIs to modern Flows without rewriting them.
- Lifecycle Awareness: Flows can be collected in coroutine scopes tied to Android lifecycles (e.g., lifecycleScope), automatically handling cancellation.
- Composability: Chain Flows with operators for transformation, error handling, etc.
- Thread Safety: Handles emissions from background threads safely.
For mid-level devs, think of it as upgrading your callbacks to a reactive pipeline. No more manual listener management—let coroutines handle it.
A Simple Example: Turning a Timer Callback into a Flow
Let’s start with a non-Android example to illustrate the mechanics. Suppose we have a hypothetical timer API that uses callbacks:
Kotlin
interface TimerCallback {
fun onTick(count: Int)
fun onFinish()
}
class Timer(private val duration: Int) {
private var callback: TimerCallback? = null
private var handler: Handler? = null
fun start(callback: TimerCallback) {
this.callback = callback
handler = Handler(Looper.getMainLooper())
var count = 0
handler?.postDelayed(object : Runnable {
override fun run() {
if (count < duration) {
callback.onTick(count)
count++
handler?.postDelayed(this, 1000)
} else {
callback.onFinish()
}
}
}, 1000)
}
fun stop() {
handler?.removeCallbacksAndMessages(null)
callback = null
}
}
Now, wrap it in a callbackFlow:
Kotlin
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.Flow
fun timerFlow(duration: Int): Flow<Int> = callbackFlow {
val timer = Timer(duration)
val callback = object : TimerCallback {
override fun onTick(count: Int) {
trySend(count) // Emit the count
}
override fun onFinish() {
close() // Complete the Flow
}
}
timer.start(callback)
awaitClose {
timer.stop() // Cleanup on cancellation
}
}
Usage in a ViewModel or Activity:
Kotlin
viewModelScope.launch {
timerFlow(5).collect { count ->
Log.d("TimerFlow", "Tick: $count")
}
}
When the scope is cancelled (e.g., activity destroyed), awaitClose ensures the timer stops. This prevents leaks and keeps things clean. Notice how we handle completion with close()—Flows can be finite or infinite.
This example is about 300 words in, but it sets the foundation. Now, let’s apply this to Android.
Real-World Android Example: Location Updates with callbackFlow
Android’s Fused Location Provider is a classic callback-based API. We’ll convert FusedLocationProviderClient.requestLocationUpdates to a Flow.
First, ensure dependencies:
groovy
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.0"
implementation "com.google.android.gms:play-services-location:21.0.1"
The Flow factory:
Kotlin
import android.content.Context
import com.google.android.gms.location.FusedLocationProviderClient
import com.google.android.gms.location.LocationCallback
import com.google.android.gms.location.LocationRequest
import com.google.android.gms.location.LocationResult
import com.google.android.gms.location.LocationServices
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.tasks.await
fun Context.locationFlow(locationRequest: LocationRequest): Flow<LocationResult> = callbackFlow {
val client: FusedLocationProviderClient = LocationServices.getFusedLocationProviderClient(this@locationFlow)
val locationCallback = object : LocationCallback() {
override fun onLocationResult(result: LocationResult) {
trySend(result) // Emit the result
}
}
// Optional: Get initial location
val initialLocation = client.lastLocation.await()
if (initialLocation != null) {
trySend(LocationResult.create(listOf(initialLocation)))
}
client.requestLocationUpdates(locationRequest, locationCallback, Looper.getMainLooper())
.addOnFailureListener { e ->
close(e) // Close with error
}
awaitClose {
client.removeLocationUpdates(locationCallback)
}
}
Key points:
- We use trySend in the callback.
- Handle errors by closing the Flow with an exception.
- awaitClose removes the updates to prevent battery drain.
- We even fetch an initial location using await() for Tasks.
In your Activity or Fragment:
Kotlin
private fun startLocationUpdates() {
val locationRequest = LocationRequest.create().apply {
priority = LocationRequest.PRIORITY_HIGH_ACCURACY
interval = 10000 // 10 seconds
}
lifecycleScope.launchWhenStarted {
applicationContext.locationFlow(locationRequest)
.catch { e -> Log.e("LocationFlow", "Error: ${e.message}") }
.collect { result ->
val location = result.lastLocation
Log.d("LocationFlow", "Lat: ${location.latitude}, Lon: ${location.longitude}")
// Update UI
}
}
}
Don’t forget permissions! Add <uses-permission android:name=”android.permission.ACCESS_FINE_LOCATION” /> and request runtime permissions.
This Flow is infinite until cancelled, perfect for ongoing updates. You can add operators like debounce(5000) to throttle emissions.
Another Android Example: Firebase Realtime Database Listener
Firebase often uses callbacks for data changes. Let’s wrap ValueEventListener into a Flow.
Dependencies: implementation “com.google.firebase:firebase-database-ktx:20.3.0”
The Flow:
Kotlin
import com.google.firebase.database.DataSnapshot
import com.google.firebase.database.DatabaseError
import com.google.firebase.database.DatabaseReference
import com.google.firebase.database.ValueEventListener
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
fun DatabaseReference.dataFlow(): Flow<DataSnapshot> = callbackFlow {
val listener = object : ValueEventListener {
override fun onDataChange(snapshot: DataSnapshot) {
trySend(snapshot)
}
override fun onCancelled(error: DatabaseError) {
close(error.toException())
}
}
addValueEventListener(listener)
awaitClose {
removeEventListener(listener)
}
}
Usage:
Kotlin
val ref: DatabaseReference = Firebase.database.reference.child("users")
viewModelScope.launch {
ref.dataFlow().collect { snapshot ->
val user = snapshot.getValue(User::class.java)
// Update state
}
}
This handles real-time updates elegantly. For single-value reads, you could use get().await() instead, but for listening, Flow is ideal.
Best Practices and Common Pitfalls
As a senior dev, I’ve seen mistakes—let’s avoid them.
Best Practices
- Always Use awaitClose: This is your cleanup hook. Forget it, and you’ll leak resources.
- Handle Errors Properly: Use close(Throwable) for failures. In collectors, use catch operator.
- Thread Considerations: Callbacks might run on background threads, but trySend is thread-safe. If needed, use flowOn(Dispatchers.IO).
- Buffering: If emissions are fast, use buffer() to handle backpressure.
- Testing: Use TestCoroutineScope and collect in tests. Mock callbacks to simulate emissions.
- Compose with Operators: E.g., map { it.locations.last() } for location Flows.
Common Pitfalls
- Suspending in Callbacks: Don’t use suspending functions inside callbacks without care—trySend is non-suspending.
- Infinite Flows: Ensure cancellation paths are clear, especially in UI components.
- Multiple Collectors: Flows are cold, so multiple collectors start separate instances. If sharing is needed, use shareIn.
- Context Leaks: Tie to lifecycle-aware scopes.
Example of sharing:
Kotlin
val sharedLocationFlow = locationFlow(locationRequest)
.shareIn(viewModelScope, started = SharingStarted.WhileSubscribed(5000), replay = 1)
This replays the last value to new collectors.
Advanced Tips: Combining with Other Coroutine Features
For mid-level devs ready to level up:
- With StateFlow: Convert your callbackFlow to StateFlow for UI state:
Kotlin
val stateFlow: StateFlow<Location?> = locationFlow(locationRequest)
.map { it.lastLocation }
.stateIn(viewModelScope, SharingStarted.Eagerly, null)
- Error Retry: Use retryWhen { cause, attempt -> attempt < 3 && cause is IOException }.
- Integration with Room or Paging: CallbackFlows can feed into PagingSource for infinite lists.
- Custom Producers: If trySend fails (channel full), handle with offer or custom logic, but usually buffer suffices.
In performance-critical apps, profile emissions—Flows are efficient but callbacks can flood.
Conclusion: Elevate Your Android Apps with callbackFlow
We’ve covered a lot: from basics of callbacks and Flows, to building callbackFlow, real Android examples like location and Firebase, best practices, and advanced tips. As a senior dev, I use callbackFlow daily to modernize codebases, making them more testable and reactive. It’s a testament to Kotlin’s elegance in handling async paradigms.
