Access Mono object inside map operation after flatmap operation

Question

I am trying to create a gateway proxy router with Spring Cloud Gateway using a custom filter. Everything is working as intended when overriding the attributes in a blocking and imperative way.

exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + newTargetURLHost + )

String variable newTargetURLHost was obtained by using:

newTarget = serviceReturnsMono.getServerMapping(id).block().getHost();

I am fairly new to Webflux, but the above line already is a code smell to me. After further reading, this is not the best approach when working with reactive. I have tried to rewrite in a more functional/reactive way but failing to get the reactive stream to emit the desired values.

Mono.just(serviceReturnsMono.getServerMapping(id))
                    .flatMap(flat -> flat)
                    .subscribeOn(Schedulers.immediate())
                    .map(server -> exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + server.getHost() ))
                    .subscribe();

When the above code executes, the exchange attribute does not get mutated.

I have also tried the following without success:

            serverMappingMono
                    .map(serverMapping -> exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + serverMapping.getHost() ))
                    .subscribe();

As a test, when I modify the code as below to troubleshoot, the following does emit a hardcoded string and the exchange attribute is mutated.

                    .map(server -> exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + "testHostName" ))
                    .subscribe();

Any ideas or pointers would be greatly appreciated.

UPDATE: Filter code as follows:


    private ReturnsMonoServerMappingService returnsMonoServerMappingService;

    @Override
    public int getOrder() {
        return 10001;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        final String id = exchange.getRequest().getHeaders().getFirst("reference");

        return chain.filter(exchange).then(Mono.fromRunnable(() -> {

            Mono<ServerMapping> serverMapping = returnsMonoServerMappingService.getServerMapping(id);
            serverMapping
                    .map(server -> exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + server.getHost()  )))
                    .subscribe();

        }));
    }
}

UPDATE SOLUTION FROM Thomas:

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

    final String id = exchange.getRequest()
                                 .getHeaders()
                                 .getFirst("reference");

    return returnsMonoServerMappingService.getServerMapping(id)
                     .doOnSuccess(serverMapping -> {                
                         exchange.getAttributes()
                             .put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + server.getHost()
        }).then(chain.filter(exchange));
}

The missing piece that I was overlooking was "doOnSuccess" as returnsMonoServerMappingService already returns a mono. Then chain the exchange by "then" to delegate to the next filter in the chain.

you should never subscribe, the subscriber is the calling client. And you are right you should never block. Please post your entire filter code and just not a snippet. — Toerktumlare
looks like you are missing the top part of your filter code... please edit — Toerktumlare
and you havn't described what it is you want to do. You want to get a header, lookup a servermapping, and then add that as an attribute to the request? — Toerktumlare

Toerktumlare Toerktumlare · Accepted Answer · 2019-07-15T01:28:51

Im writing this on my phone, so can't test it and im writing from memory but it should be something like this i think. Or at least you get the gist.

We first extract the id. Then we lookup the servermapping, and if that goes well we put it as an attribute, then after that we continue the filter chain.

You should almost never subscribe in an application, the calling client is usually the subscriber. And never block in a reactive application.

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

    final String id = exchange.getRequest()
                                 .getHeaders()
                                 .getFirst("reference");
    
    return returnsMonoServerMappingService.getServerMapping(id)
                     .doOnSuccess(serverMapping -> {                
                         exchange.getAttributes()
                             .put(GATEWAY_REQUEST_URL_ATTR, URI.create("https://" + server.getHost()
        }).then(chain.filter(exchange));
}

Access Mono object inside map operation after flatmap operation

1 Answers