支持通过Spring Cloud Gateway代理SOAP/XMLendpoint

Question

尝试在使用Spring cloud Gateway代理soap服务时记录请求和响应。下面是代码段...它引发IllegalStateException。

IllegalStateException：只允许一个连接接收订阅服务器。在reactor.netty.channel.fluxreceive.startreceive(fluxreceive.java:271)[reactor-nety.channel.fluxreceive.subscribe(fluxreceive.java:121)[reactor-nety.channel.fluxreceive.jar:0.8.4.release.jar:0.8.4.release.jar:0.8.4.release]

--- Spring Versions ---

SpringBoot                  : 2.1.2.RELEASE
spring-cloud-gateway        : 2.1.1.BUILD-SNAPSHOT
spring-boot-starter-webflux : 2.1.2.RELEASE
reactor-core                : 3.2.5.RELEASE

GatewaySampleApplication

public class  GatewaySampleApplication {
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
return builder.routes() 
.route(
r ->
r.path("/security/getToken")
.filters( f -> f.preserveHostHeader()
.filter(new PolicyHandlerReactive()))
.uri("http://targetendpoint.xyz.com")
)
.build();
}
}

PolicyHandlerReactive：

PolicyHandlerReactive implements GatewayFilter  {

@Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) 
    {
        ServerHttpRequest request = (ServerHttpRequest) exchange.getRequest();
                Flux<DataBuffer> xmlReqFlux = request.getBody();    
                Mono<List<DataBuffer>> monoDataList = xmlReqFlux.collectList();
                monoDataList.map(dataA -> logRequest2(dataA)).subscribe();
        return chain.filter(exchange);
        }
}

private List<DataBuffer> logRequest(List<DataBuffer> dataA) {           

    ListIterator<DataBuffer> dataAList = dataA.listIterator();

    while(dataAList.hasNext() ) {
        DataBuffer dataBuffer = dataAList.next();
        InputStream is = dataBuffer.asInputStream();
        System.out.println(convertStreamToString(is));
    }       
    return dataA;
}   

Answer 1

能够使用.ModifyRequestBody而不是筛选器解决此问题...更新了以下java DSL...

    .route(
                    r ->
                    r.path("external/v2/security/")
                    .filters( f -> f
                            .preserveHostHeader()
                            .modifyRequestBody(String.class, String.class, 
                            (exchange, s) -> {
                                Mono<String> dataA = Mono.just(s); 
                                dataA.subscribe(System.out::print); 
                                return dataA; 
                            })                              
                         )                                                                                               
                    .uri("http://targetendpoint.xyz.com")
                    )