Build Flow result based on call to a suspend function

user_386201 0
Viewed 1581

I am learning coroutines and need some help to understand a basic use case.

Implement a non-blocking method that:

  • Fetches a single item from a (reactive) DB
  • Determines a range (i.e. the month that the item lives in) based on that item's timestamp
  • Fetches all items in that month
  • Returns the items as Flow

Approach

Because it must return a Flow I will not use suspend (like I would when returning a single item). Returning Flow and suspend (which kind of returns a Mono) are most commonly mutually exclusive, right?

So I came up with this signature:

override fun getHistory(beforeUtcMillisExclusive: Long): Flow<Item>

Trying an implementation:

val itemInNextPeriod = itemRepository.findOneByTimestampLessThan(beforeUtcMillisExclusive)
if (itemInNextPeriod == null) {
  return emptyFlow()
} else {
  val range = calcRange(itemInNextPeriod.timestamp)
  return itemRepository.findByTimestampGreaterThanEqualAndTimestampLessThan(range.start, range.end)
}

This gives me on the very first line:

Suspend function 'findOneByTimestampLessThan' should be called only from a coroutine or another suspend function

I understand the problem that we are not allowed to call a suspend function here and the proposed solution by IntelliJ "adding suspend" does not make sense, when already returning a flow.

So, from this question I got the idea of using a return flow {...}:

return flow {
  val itemInNextPeriod = itemRepository.findOneByTimestampLessThan(beforeUtcMillisExclusive)
  if (itemInNextPeriod == null) {
    return@flow
  } else {
    val range = calcRange(itemInNextPeriod.timestamp)
    return@flow itemRepository.findByTimestampGreaterThanEqualAndTimestampLessThan(range.start, 
  range.end)
  }
}

The second repository call findByTimestampGreaterThanEqualAndTimestampLessThan returns Flow<Item> and I do not understand why I cannot return it.

This function must return a value of type Unit Type mismatch. Required: Unit Found: Flow

kotlin-coroutines kotlin
1 Answers

return@flow returns from the lambda, not from enclosing function. You need to reemit items from Flow returned by findByTimestampGreaterThanEqualAndTimestampLessThan call into Flow you're building with flow function:

return flow {
    val itemInNextPeriod = itemRepository.findOneByTimestampLessThan(beforeUtcMillisExclusive)
    if (itemInNextPeriod != null) {
        val range = calcRange(itemInNextPeriod.timestamp)
        emitAll(itemRepository.findByTimestampGreaterThanEqualAndTimestampLessThan(range.start, range.end))
    }
}
user_139686730
Related