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)
                }
            }
}

Answer 2

You can check it using the Mono's provided method hasElement() which is analogous to Optional's isPresent(). The method definition is :

Mono<Boolean> hasElement()

for more details checkout : project reactor documentation

In case you have to perform some action based on this value you can further use switchIfEmpty() to provide with alternate publisher.

Answer 3

Use Mono with Optional:

return findExistingUserMono
  .map(Optional::of)
  .defaultIfEmpty(Optional.empty())
  .flatMap(optionalUser -> {
    if(optionalUser.isPresent()) {
      return Mono.error('xxxx');
    }
    return this.userService.create(optionalUser.get());
  });

This way it will always emit Optional value so that the stream will never break.

Answer 4

Let me start by saying I am a newbie on reactive (java) and on this forum. I think you cannot really check in this code if a mono is empty because a mono represents code that will be executed later on, so in this code body you won't know yet if its is empty. Does that make sense?

I just wrote something similar in Java which seems to work (but not 100% this is the best approach either):

    public Mono<ServerResponse> queryStore(ServerRequest request) { 

        Optional<String> postalCode = request.queryParam("postalCode");                            

        Mono<ServerResponse> badQuery = ServerResponse.badRequest().build();
        Mono<ServerResponse> notFound = ServerResponse.notFound().build();

        if (!postalCode.isPresent()) { return  badQuery; }

        Flux<Store> stores = this.repository
                .getNearByStores(postalCode.get(), 5);

        return ServerResponse.ok().contentType(APPLICATION_JSON)
                .body(stores, Store.class)
                .switchIfEmpty(notFound);
}

Answer 5

We can use switchIfEmpty method for this

Below example, I'm checking if the user exists with email if not then add it

userRepository.findByEmail(user.getEmail())
                .switchIfEmpty(s -> {
                    user.setStatus("InActive");
                    String encodedPassword = DigestUtils.sha256Hex(user.getPassword());
                    user.setPassword(encodedPassword);
                    userRepository.save(user).subscribe();
                    s.onComplete();
                }).then(Mono.just(user));