I'm stress testing some callbackFlow code I have written. Here it is currently

fun getBooks(): Flow<List<Book>> {
  return callbackFlow {
    val listener =
      FirebaseFirestore.getInstance()
        .collection("books")
        .addSnapshotListener { value, error ->
          if (error != null || value == null) { /*dosomething*/ }
          var books: List<Book>? = null
          runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
              books = (value!!.toObjects())
          }
          trySend(books!!)
        }
    awaitClose { listener.remove() }
  }
}

this seems to work well... but to stress test it I wrapped

books = (value!!.toObjects())

with

repeat (1000)

and now my UI hangs. Shouldn't the dispatchers.io take care of this?

runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
  repeat(1000) {
    books = (value!!.toObjects())
  }
}

Dispatchers.IO schedules to the background all right. But runBlocking blocks your thread waiting for it to be done. If being asynchronous is OK for you, you can use `launch`instead of

runBlocking

If you need to wait for

books

to be assigned, you have an issue. Outside couroutines, you can not wait without blocking your current thread.

Hm. I was trying to use repeat because I'm trying to stress test whether or not toObject() takes a long time. Hence me doing repeat(1000)

So, what is your question?

You answered my original question. I was just trying to respond to your suggestion of just using

scope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {
  send(value!!.toObject)
}

This isn’t going to handle backpressure well and may lead to events being emitted out of order or other bad things

you should also call this with

flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

interesting

check this for context in flows https://elizarov.medium.com/execution-context-of-kotlin-flows-b8c151c9309b

fun getBooks(): Flow<List<Book>> {
  return callbackFlow {
    val listener =
      FirebaseFirestore.getInstance()
        .collection("books")
        .addSnapshotListener { value, error ->
          if (error != null) { /*dosomething*/ }
          trySend(value)
        }
    awaitClose { listener.remove() }
  }
  .filterNotNull()
  .map{
    value.toObjects()
  }
  .flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

@Francesc, @Colton Idle I guess, that’s a about the way

flowOn is not needed in the Google code because the Flow is a simple Channel backed flow so there’s nothing to run off the main thread

it was meant for the

value.toObjects()

Hence it’s important to return as soon as possible and do anything expensive on the Flow

so the goal is to get the

map

off the main thread

So does everyones generally agree that @uli's last code snippet is what i should be doing?

Perfect time to close this thread as my train arrives in one minute ;-)

that works, but if your collector is slow you may drop emissions

if my collector is slow... i may drop emissions. pardon my ignorance. but is this "backpressure"?

yes, you have a

trySend

so it will try, but if the buffer is full, it will drop it

Yeah. callback flow has a fixed buffer so this is unavoidable when you don’t control the source

Just add

.conflate()

after

callbackFlow

so that it always just keeps the latest value from firestore, I assume that's the only one you care about. And it looks like you can ditch

.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

, I don't see any blocking code

@Nick Allen https://kotlinlang.slack.com/archives/C1CFAFJSK/p1659543557555079?thread_ts=1659541071.357849&cid=C1CFAFJSK

Yep,

conflate

and/or replace

map

with

mapLatest

ooh. Map latest...

They do slightly different things and you can actually use both

both... at the same time?

conflate reduces buffer size to 1 and only keeps latest

Do not use

mapLatest

for back-pressure. Only use it if

mapLatest

lambda is suspending code that actually suspends (not blocking on IO dispatcher). It processes every item and when it gets a new item it cancels and waits for the previous lambda (just the lambda, nothing downstream). If the code after mapLatest is slow, then you'll still have backpressure and could miss updates. They are completely separate and you should not interchange them

Okay. it looks like my end result is...

fun getBooks(): Flow<List<Book>> {
  return callbackFlow {
    val listener =
      FirebaseFirestore.getInstance()
        .collection("books")
        .addSnapshotListener { value, error ->
          if (error != null) { /*dosomething*/ }
          trySend(value)
        }
    awaitClose { listener.remove() }
  }
  .conflate()
  .filterNotNull()
  .map{
    value.toObjects()
  }
  .flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

mapLatest

does work better if the block supports cooperative cancellation but it won’t caused missed updates any more than conflate will

looks good. A thing to consider is that you should not hardcode dispatchers, consider using a class that provides a wrapper for the dispatchers so that you can then replace those for tests

Unless the producing source is constantly faster than the mapping in which case it’ll never be able to emit anything

If

mapLatest

is used with a block that doesn't support cooperative cancellation (like deserialization), then every new event is blocked waiting for the previous lambda invocation to finish, and every item is processed, which could result in backpressure causing trySend to fail and so the most recent value could be dropped. This is the opposite of what you want, which is to keep the latest item.

Next train … Actually I left error handling in your callback. Whatever ‘dosomething’ is should probably also go into the stream:

trySend(value to error)

@Nick Allen You’re right, for some reason I was thinking it’d still drop the final output but I guess it’s actually identical to map if your block can’t cancel

oooh that last snippet @uli looks like it ticks all of the boxes

I just update some of my own code, a shared preferences listener. It received a key and an instance of the shared preferences. And it used to load the value in the callback. I do not think that this ever was an issue, as shared preferences are usually read from memory. But I feel the pattern of minimal work in the callback is worth following just as a habit. No reason to ever question it.

Sorry this bring this conversation up again, but I essentially have 10 functions now that are pretty much identical to the above (since I have like 10 different queries in my app). If I want to make a generic function that could do this, I basically came up with this. Thoughts?

fun <T> firestoreFlow(query: Query, clazz: Class<T>, dispatcher: CoroutineDispatcher): Flow<List<T>> {
  return callbackFlow {
        val listener = query.addSnapshotListener { value, _ -> trySend(value) }
        awaitClose { listener.remove() }
      }
      .conflate()
      .filterNotNull()
      .map { value -> value.toObjects(clazz) }
      .flowOn(dispatcher)
}

I would remove the dispatcher from your method and let the caller add the

flowOn

. I would also pass a lambda that does the mapping for you.

Interesting. yeah. I can see how those things could be passed in. I guess I'm trying to find the right level of abstraction and convenience for whoever has to come into my ApiService.kt and add a new flow. I want them to have an easy way to not mess this up (as flowables are still new to our team)

I see, what you had is more rigid but if that's all you need, it's easier to use

yeah. i think that's what im going to do. essentially one could be shared online in a gist and anyone in the world could use (so to speak) and then I can have one that's a convenience for my codebase that builds on top of that. awesome. This has me so excited. I've learned so much the past few days. I think the only thing I'm still a bit uneasy about is how to handle errors and stuff, but maybe firestoer doesn't really send down errors if your query is correct. /shruggie

you can propagate the error down the flow, you can wrap the error in your custom exception and throw it, then whoever is collecting your flow can add a

.catch

block to the flow and decide how to handle the error. The collector should have a much better context to be able to decide what to do with the error

Pardon my ignorance here... but how would I do that? Now that you saw it. Putting the "error handling" on the caller does make a lot more sense. But I'm not sure how that changes my observable and what that looks like on the observer.

give me a minute to put together a gist

To send an exception to the flow, you need to call

close

with the exception instead of throwing it.

thanks for pointing that out

Does an exception/"close" terminate the flow? I would think that I would want the flow to continue even if there's one error that's thrown in there.

You can use retry for that

thanks. im going to try a bit to force some failures just to see what its all about.

One more tip: you often can avoid passing Class in kotlin by using inline and reified https://kotlinlang.org/docs/inline-functions.html#reified-type-parameters

Looks like firestore IS getting support for Flows! Time to see how they implemented it! https://github.com/firebase/firebase-android-sdk/pull/1252#issuecomment-1213179027