block()/blocking() not supported in thread reactor

Question

Here is a logic to fetch some information from an API:

class OneClient {
    public Mono<Information> getSomeInformation(ClientRequest req) {
        return getWebTarget(req)
            .request(MediaType.APPLICATION_JSON_TYPE)
            .headers(getHeaders(req))
            .rx(MonoRxInvoker.class)
            .get()
            .map(this::processResponse)
            .doOnError(this::processError);
    }
}

Thats how our framework is to be used by putting the request on a different thread while keeping main thread unblocked.

But I have a problem. getHeaders() gets information like below:

public MultiValuedMap<String,Object> getHeaders() {
    if(token == null) {
        getTokenFromExternalApi() // just like previous snippet
        .map(resp -> setToken(resp))
        .block();
    }
    return headers; // tokens and other static data
}

Since I have to store token in the bean, I was doing .block(), but then I get exception: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-5. And I understand why this error is being thrown. Blocking is unexpected in a non-blocking thread.

But

Can you suggest me the right way to design getHeader()? Thank you.

Answer 1

I apologize for my earlier response not fully addressing your issue.

you can modify getHeaders() to return a Mono that emits the headers object when the token is available. You can then chain this Mono with the map() operator to set the headers and continue with the request execution.

public Mono<MultiValuedMap<String,Object>> getHeaders() {
   if(token == null) {
    return getTokenFromExternalApi()
        .map(resp -> setToken(resp))
        .then(Mono.just(headers)); // return headers after setting token
}
return Mono.just(headers); // return headers if token is already set
}

To use this implementation, you can modify the webTarget code as follows:

getHeaders()
.map(headers -> {
    webTarget.headers(headers);
    return webTarget;
})
.flatMap(webTarget -> webTarget.rx(MonoRxInvoker.class).get())
.map(this::processResponse)
.doOnError(this::processError);