is there a better way to check if a flow is emtpy ...

# coroutines

[Dsittel]

is there a better way to check if a flow is emtpy then,

flow.count() == 0 ?

[gildor]

Flow is lazy and cold, there is no way to know how many elements are there until you start consuming it and consume all elements, so there is no other way, it in general look as not a correct approach

[gildor]

also this code may never return, it’s valid for flow never send complete/error

[gildor]

what is your use case for it?

[Dsittel]

have a connection from webflux (spring) to cosmos db and this returns a Flow<T> but if there are no elements i want to throw a custum exception from the backend and not an emtpy list (cause i am returning an APPLICATION_JSON)

[gildor]

but if there are no elements

But Flow<T> from database is not a container for list of elements, it probably a stream of updates of database

[gildor]

if you want to retrive current state of DB, use Flow as return type is not the best solution, it should be a suspend function

[Dsittel]

like suspend fun : List<T> instead of fun : Flow<T> ?

[Dsittel]

all docs say to use Flux for > 1 element for a repository as i ve seen so far

[gildor]

yes, like suspend for List

[gildor]

I really not sure how this Flow works, it’s asyncronous, will it ever complete?

[Dsittel]

yeah ll complete

[Dsittel]

count() is a collector (is also a suspend function)

[Dsittel]

its working so far but maybe a overkill here tbh 🙂

[gildor]

if it does, it’s quite strange, but then you should do just this:
val t = flow.toList()
t.size() == 0

[gildor]

It’s not overkill, I just think such API is incorrect. to expose Flow<T> if actually it represent List<T>

[Dsittel]

https://spring.io/blog/2019/04/12/going-reactive-with-spring-coroutines-and-kotlin-flow

[Dsittel]

same pattern

[gildor]

Shouldn’t in webflux you do this:
select().blah.fetch().awaitOneOrNull()

[gildor]

or just await(), for liost

[Dsittel]

@GetMapping("/")
	fun findAll(): Flow<User> =
		userRepository.findAll()

[Dsittel]

yeah i am working with await for single events

[Dsittel]

replacement for Mono from reactor package

[Dsittel]

we are pretty new as a team for all this topics but we have much fun :=

[gildor]

so it looks that webflux really asyncronously request all items

[gildor]

inteeresting

[gildor]

findAll().toList()

[gildor]

is essentially what you need

[gildor]

but I still not sure why the do not have Mono<List<T>> together with suspend List<T> instead of Flux<T> and Flow<T>

[gildor]

because essentially, Flow or Flux as result doesn’t work for you case, you have to return whole result in one JSON

[gildor]

so I really don’t see other solution except
findAll().toList()

[Dsittel]

yeah this is exactly what happens under the hood => return as whole json

[gildor]

yeah, but I quite surprised that such explicit behaviour is possible to achive only implicitly assuming how Flow<T> or Flux<T> is implemented under the hood

[Dsittel]

so far its working quite nice

[Dsittel]

the count() == 0 feels ugly but maybe i have to return an empty list and frontend can do the differeniation

[gildor]

my problem with count() == 0
is that you anyway need data, count() consumes flow, so if you do this, it would be incorrect:

fun getData(): Flow<T> {
  require(flow.count() > 0) { “DB is empty }
  // at this monent DB is changed, items deleted
  return flow // you still have to request flow again
}
```

[gildor]

so maybe this will work for you:

return flow.onEmpty { throw DbException(“DB is empty”) }

[gildor]

so consumer will receive exception if flow is empty

[Dsittel]

cool i am gonna try this approach

[christophsturm]

whats the downside of always converting to a list and throwing if its empty?

[gildor]

Yeah, I think it what you need exactly, and very reactive, but still a bit strange for me, suspend List<T> is what actually needed and it has nothing to do with blocking code

[gildor]

Downside is that you will change signature of the method

[gildor]

which may be a problem if conumer expects Flow<T>

[gildor]

and does some processing on each element
As it pointed out on this article from Spring blog, it can be a problem for processing, to load all data to memory

[gildor]

but if only what consumer does is get result and convert to Json, then from API point of view I would rather use suspend List

[gildor]

Also it depends on how Webflux implemented, maybe such way to receive every item of result asyncronousle is more optimal solution, they somehow parallel it and it supported by DB driver

[christophsturm]

yeah, if the normal path is converting it to a list at some point thats where it should throw

[gildor]

I’m just worry about such approach in general, you received some Flow<T>, if your code has no option as get all items, it becomes dangerous, if someone changes upstream implementation and it will never complete
In such cases more safe probably to have some timeouts when you collecting data to avoid forever suspended code, which, same as deadlock, is very nasty bug and hard to debug

So provide a clear API to receive List of data looks as good solution, and let user decide, use List<T> or Flow<T> depending on case

[Luis Munoz]

I think you want to return Flow<User> because some database drivers can read 1 row at a time lazily

[gildor]

yes, this what I meant by optimisations, that it supported on level of DB driver, it actually pretty cool, but if you application sends all results together, something like HTTP, and not some streaming socket, this lazy select will not help and probably will be a bit slower

[Dsittel]

true Andrey

[christophsturm]

in htttp you can stream the reply and even if you don’t you can start to build the reply while data is still coming in.

[gildor]

which not necessary always possible (like case an error for 0 elements) and not always faster, especially if building the reply is not heavy operation

[christophsturm]

even if you just want to build a json string from a list of replies you can avoid creating an immediate List

[gildor]

it’s true

[gildor]

until you have something like count property in your json