Java Reactive stream how to map an object when the object being mapped is also needed on the next step of the stream

Question

I am using Java 11 and project Reactor (from Spring). I need to make a http call to a rest api (I can only make it once in the whole flow). With the response I need to compute two things:

1. Check if a document exists in the database (mongodb). If it does not exists then create it and return it. Otherwise just return it.
2. Compute some logic on the response and we are done.

In pseudo code it is something like this:

public void computeData(String id) {
    httpClient.getData(id) // Returns a Mono<Data>
        .flatMap(data -> getDocument(data.getDocumenId()))
         // Issue here is we need access to the data object consumed in the previous flatMap but at the same time we also need the document object we get from the previous flatMap
        .flatMap(document -> calculateValue(document, data)) 
        .subscribe();
}

public Mono<Document> getDocument(String id) {
    // Check if document exists
    // If not create document

    return document;
}

public Mono<Value> calculateValue(Document doc, Data data) {
    // Do something...
    return value;
}

The issue is that calculateValue needs the return value from http.getData but this was already consumed on the first flatMap but we also need the document object we get from the previous flatMap.

I tried to solve this issue using Mono.zip like below:

public void computeData(String id) {
    final Mono<Data> dataMono = httpClient.getData(id);

    Mono.zip(
        new Mono<Mono<Document>>() {
            @Override
            public void subscribe(CoreSubscriber<? super Mono<Document>> actual) {
                final Mono<Document> documentMono = dataMono.flatMap(data -> getDocument(data.getDocumentId()))
                actual.onNext(documentMono);
            }
        },
        new Mono<Mono<Value>>() {
            @Override
            public void subscribe(CoreSubscriber<? super Mono<Value>> actual) {
                actual.onNext(dataMono);
            }
        }
    )
    .flatMap(objects -> {
        final Mono<Document> documentMono = objects.getT1();
        final Mono<Data> dataMono = objects.getT2();

        return Mono.zip(documentMono, dataMono, (document, data) -> calculateValue(document, data))
    })
}

But this is executing the httpClient.getData(id) twice which goes against my constrain of only calling it once. I understand why it is being executed twice (I subscribe to it twice).

Maybe my solution design can be improved somewhere but I do not see where. To me this sounds like a "normal" issue when designing reactive code but I could not find a suitable solution to it so far.

My question is, how can accomplish this flow in a reactive and non blocking way and only making one call to the rest api?

PS; I could add all the logic inside one single map but that would force me to subscribe to one of the Mono inside the map which is not recommended and I want to avoid following this approach.

EDIT regarding @caco3 comment I need to subscribe inside the map because both getDocument and calculateValue methods return a Mono.

So, if I wanted to put all the logic inside one single map it would be something like:

public void computeData(String id) {
    httpClient.getData(id)
        .map(data -> getDocument(data).subscribe(s -> calculateValue(s, data)))
        .subscribe();
}

Answer 1

Boiling it down, your problem is analogous to:

Mono.just(1)
        .flatMap(original -> process(original))
        .flatMap(processed -> I need access to the original value and the processed value!
                System.out.println(original); //Won't work
        );


private static Mono<String> process(int in) {
    return Mono.just(in + " is an integer").delayElement(Duration.ofSeconds(2));
}

(Silly example, I know.)

The problem is that map() (and by extension, flatMap()) are transformations - you get access to the new value, and the old one goes away. So in your second flatMap() call, you've got access to 1 is an integer, but not the original value (1.)

The solution here is to, instead of mapping to the new value, map to some kind of merged result that contains both the original and new values. Reactor provides a built in type for that - a Tuple. So editing our original example, we'd have:

Mono.just(1)
        .flatMap(original -> operation(original))
        .flatMap(processed -> //Help - I need access to the original value and the processed value!
                System.out.println(processed.getT1()); //Original
                System.out.println(processed.getT2()); //Processed

                ///etc.
        );


private static Mono<Tuple2<Integer, String>> operation(int in) {
    return Mono.just(in + " is an integer").delayElement(Duration.ofSeconds(2))
            .map(newValue -> Tuples.of(in, newValue));
}

You can use the same strategy to "hold on" to both document and data - no need for inner subscribes or anything of the sort :-)

Answer 2

You do not have to subscribe inside map, just continue building the reactive chain inside the flatMap:

getData(id) // Mono<Data>
        .flatMap(data -> getDocument(data.getDocumentId()) // Mono<Document>
                .switchIfEmpty(createDocument(data.getDocumentId())) // Mono<Document>
                .flatMap(document -> calculateValue(document, data)) // Mono<Value>
         )
        .subscribe()