spring-webflow - 支持通过 Spring Cloud Gateway 代理 SOAP/XML 端点

Question

尝试在使用 Spring 云网关代理 soap 服务时记录请求和响应。下面是代码片段...抛出 IllegalStateException。

java.lang.IllegalStateException: Only one connection receive subscriber allowed. at reactor.netty.channel.FluxReceive.startReceiver(FluxReceive.java:271) [reactor-netty-0.8.4.RELEASE.jar:0.8.4.RELEASE] at reactor.netty.channel.FluxReceive.subscribe(FluxReceive.java:121) [reactor-netty-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

网关示例应用程序

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")
                    )