Android Kotlin Coroutines: what is the difference between flow, callbackFlow, channelFlow,... other flow constructors

user_4415642 0
Viewed 14524

I have code that should change SharedPreferences into obsarvable storage with flow so I've code like this 

internal val onKeyValueChange: Flow<String> = channelFlow {
        val callback = SharedPreferences.OnSharedPreferenceChangeListener { _, key ->
            coroutineScope.launch {
                //send(key)
                offer(key)
            }
        }

        sharedPreferences.registerOnSharedPreferenceChangeListener(callback)

        awaitClose {
            sharedPreferences.unregisterOnSharedPreferenceChangeListener(callback)
        }
    }

or this 

internal val onKeyValueChange: Flow<String> = callbackFlow {
        val callback = SharedPreferences.OnSharedPreferenceChangeListener { _, key ->
            coroutineScope.launch {
                send(key)
                //offer(key)
            }
        }

        sharedPreferences.registerOnSharedPreferenceChangeListener(callback)

        awaitClose {
            sharedPreferences.unregisterOnSharedPreferenceChangeListener(callback)
        }
    }

Then I observe this preferences for token, userId, companyId and then log into but there is odd thing as I need to build app three times like changing token not causes tokenFlow to emit anything, then second time new userId not causes userIdFlow to emit anything, then after 3rd login I can logout/login and it works. On logout I am clearing all 3 properties stores in prefs token, userId, companyId.

kotlin-flow kotlin-coroutines kotlin
1 Answers

For callbackFlow:

You cannot use emit() as the simple Flow (because it's a suspend function) inside a callback. Therefore the callbackFlow offers you a synchronized way to do it with the trySend() option.

Example:

fun observeData() = flow {
 myAwesomeInterface.addListener{ result ->
   emit(result) // NOT ALLOWED
 }
}

So, coroutines offer you the option of callbackFlow:

fun observeData() = callbackFlow {
 myAwesomeInterface.addListener{ result ->
   trySend(result) // ALLOWED
 }
 awaitClose{ myAwesomeInterface.removeListener() }
}

For channelFlow:

The main difference with it and the basic Flow is described in the documentation:

A channel with the default buffer size is used. Use the buffer operator on the resulting flow to specify a user-defined value and to control what happens when data is produced faster than consumed, i.e. to control the back-pressure behavior.

The trySend() still stands for the same thing. It's just a synchronized way (a non suspending way) for emit() or send()

I suggest you to check Romans Elizarov blog for more detailed information especially this post.

Regarding your code, for callbackFlow you wont' be needing a coroutine launch:

coroutineScope.launch {
                send(key)
                //trySend(key)
            }

Just use trySend()

user_89143360
Related