Ответ 1

Методы, позволяющие проверить, является ли Flux/Mono пустым

Использование операторов .switchIfEmpty/.defaultIfEmpty/Mono.repeatWhenEmpty

Используя упомянутые операторы, вы сможете отреагировать на случай, когда Stream завершен, без выделения каких-либо элементов.

Прежде всего, помните, что такие операторы, как .map, .flatMap, .filter и многие другие, вообще не будут вызываться, если не было вызвано onNext. Это означает, что в вашем случае следующий код

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

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

не будет вызван вообще, если transaction будет пустой.

В случае, если есть требование для обработки случаев, когда ваш поток пуст, вы должны рассмотреть следующие операторы следующим образом:

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

Актуальное решение

Также я отметил, что вы создали два подпотока из основной transaction. На самом деле, следующий код не будет выполнен вообще:

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

и будет выполняться только последний, который возвращается из метода. Это происходит потому, что вы не подписаны с помощью оператора .subscribe(...).

Второй момент: вы не можете подписаться на одно и то же тело запроса больше одного раза (своего рода ограничение для ответа WebClient). Таким образом, вы должны поделиться своим телом запроса следующим образом, поэтому завершенный пример будет:

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

Или более простой случай без разделения потока транзакций, но с использованием 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)
                }
            }
}

Ответ 2

Позвольте мне начать с того, что я новичок в реактивной (java) и на этом форуме. Я думаю, вы не можете проверить этот код, если моно пусто, потому что моно представляет собой код, который будет выполнен позже, поэтому в этом теле кода вы еще не узнаете, если он пуст. Имеет ли это смысл?

Я просто написал что-то подобное на Java, которое, похоже, работает (но не на 100% это лучший подход):

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