Spring Reactive WebClient

Question

I have a reactive rest api (webflux), also using the spring WebClient class, to request data from other rest services.

Simplified design:

@PostMapping(value = "/document")
public Mono<Document> save(@RequestBody Mono<Document> document){

//1st Problem: I do not know how to get the documentoOwner ID 
//that is inside the Document class from the request body without using .block()
    Mono<DocumentOwner> documentOwner = documentOwnerWebClient()
       .get().uri("/document-owner/{id}", document.getDocumentOwner().getId())
       .accept(MediaType.APPLICATION_STREAM_JSON)
       .exchange()
       .flatMap(do -> do.bodyToMono(DocumentOwner.class));

    //2nd Problem: I need to check (validate) if the documentOwner object is "active", for instance
    //documentOwner and document instances below should be the object per se, not the Mono returned from the external API
    if (!documentOwner.isActive) throw SomeBusinessException();

    document.setDocumentOwner(documentOwner);

    //Now I can save the document in some reactive repository, 
    //and return the one saved with the given ID.
    return documentRepository.save(document)

}

In other words: I understand (almost) all of the reactive examples individually, but I am not able to put it all together and build a simple use case (get -> validate -> save -> return) without blocking the objects.

---

The closer I could get is:

@PostMapping(value = "/document")
    public Mono<Document> salvar(@RequestBody Mono<Document> documentRequest){

        return documentRequest
                .transform(this::getDocumentOwner)
                .transform(this::validateDocumentOwner)
                .and(documentRequest, this::setDocumentOwner)
                .transform(this::saveDocument);
    }

Auxiliar methods are:

private Mono<DocumentOwner> getDocumentOwner(Mono<Document> document) {
        return document.flatMap(p -> documentOwnerConsumer.getDocumentOwner(p.getDocumentOwnerId()));
    }

    private Mono<DocumentOwner> validateDocumentOwner(Mono<DocumentOwner> documentOwner) {

        return documentOwner.flatMap(do -> {
            if (do.getActive()) {
                return Mono.error(new BusinessException("Document Owner is Inactive"));
            }
            return Mono.just(do);
        });


    }

    private DocumentOwnersetDocumentOwner(DocumentOwner documentOwner, Document document) {
        document.setDocumentOwner(documentOwner);
        return document;
    }

    private Mono<Document> saveDocument(Mono<Document> documentMono) {
        return documentMono.flatMap(documentRepository::save);
    }

I am using Netty, SpringBoot, Spring WebFlux and Reactive Mongo Repository. But there are some problems:

1) I am getting the error: java.lang.IllegalStateException: Only one connection receive subscriber allowed. Maybe because I am using the same documentRequest to transform and to setDocumentOwner. I really don't know.

2) setDocumentOwner method is not being called. So the document object to be saved is not updated. I believe could have a better way to implement this setDocumentOwner().

Thanks

Answer 1

I don't really get the point with the validation aspect of that question. But it looks like you're trying to compose reactive types together. This is something that reactor handles perfectly with operators.

There are certainly better ways to handler that case, but a quick search in the Mono API makes me think about this:

Mono<Document> document = ...
Mono<DocumentOwner> docOwner = ...
another = Mono.when(document, docOwner)
              .map(tuple -> {
                        Document doc = tuple.getT1();
                        DocumentOwner owner = tuple.getT2();
                        if(owner.getActive()) {
                          return Mono.error(new BusinessException("Document Owner is Inactive"));
                        }
                        doc.setDocumentOwner(owner);
                        return doc;
              })
              .flatMap(documentRepository::save);