How to check if Mono is empty?

Question

I'm developing a app with Spring Boot 2.0 and Kotlin using the WebFlux framework.

I want to check if a user id exits before save a transaction. I'm stucked in a simple thing like validate if a Mono is empty.

fun createTransaction(serverRequest: ServerRequest) : Mono<ServerResponse> {
    val transaction = serverRequest.body(BodyExtractors.toMono(Transaction::class.java))

    transaction.flatMap {
        val user = userRepository.findById(it.userId)
        // If it's empty, return badRequest() 
    } 

    return transaction.flatMap { transactionRepository.save(it).then(created(URI.create("/transaction/" + it.id)).build()) }
}

It is possible to do what I want?

Answer 1

The techniques that allow checking whether Flux/Mono is empty

Using operators .switchIfEmpty/.defaultIfEmpty/Mono.repeatWhenEmpty

Using mentioned operators you will be able to react to the case when Stream has been completed without emitting any elements.

First of all, remember that operators such .map, .flatMap, .filter and many others will not be invoked at all if there no onNext has been invoked. That means that in your case next code

transaction.flatMap {
    val user = userRepository.findById(it.userId)
    // If it's empty, return badRequest() 
} 

return transaction.flatMap { transactionRepository.save(it).then(created(URI.create("/transaction/" + it.id)).build()) }

will not be invoked at all, if transaction will be empty.

In case if there is a requirement for handling cases when your flow is empty, you should consider operators like next in the following manner:

transaction
   .flatMap(it -> {
      val user = userRepository.findById(it.userId)
   })
   .swithIfEmpty(Flux.defer(() -> Flux.just(badRequest())));

Actual solution

Also, I have noted that you created two sub-flows from the main transaction. Actually, following code will not be executed at all:

transaction.flatMap {
    val user = userRepository.findById(it.userId)
    // If it's empty, return badRequest() 
}  

and will be only executed the last one, which is returned from the method. That happens because you ain't subscribed using operator .subscribe(...).

The second point, you can't subscribe to the same request body more the one time (kind of limitation for WebClient's reponse). Thus you are required to share your request body in the next way, so completed example will be:

fun createTransaction(serverRequest: ServerRequest): Mono<ServerResponse> {
    val transaction = serverRequest.body(BodyExtractors.toMono(Transaction::class.java)).cache()

    transaction
            .flatMap { userRepository.findById(it.userId) }
            .flatMap { transaction.flatMap { transactionRepository.save(it) } }
            .flatMap { ServerResponse.created(URI.create("/transaction/" + it.id)).build() }
            .switchIfEmpty(transaction.flatMap { ServerResponse.badRequest().syncBody("missed User for transaction " + it.id) })
}

Or more simple case without sharing transaction flow but using Tuple:

fun createTransaction(serverRequest: ServerRequest): Mono<ServerResponse> {
    val emptyUser = !User()
    val transaction = serverRequest.body<Mono<Transaction>>(BodyExtractors.toMono(Transaction::class.java))

    transaction
            .flatMap { t ->
                userRepository.findById(t.userId)
                        .map { Tuples.of(t, it) }
                        .defaultIfEmpty(Tuples.of(t, emptyUser))
            }
            .flatMap {
                if (it.t2 != emptyUser) {
                    transactionRepository.save(it.t1)
                            .flatMap { ServerResponse.created(URI.create("/transaction/" + it.id)).build() }
                } else {
                    ServerResponse.badRequest().syncBody("missed User for transaction " + it.t1.id)
                }
            }
}