尝试使用Spring云网关代理soap服务时记录请求和响应。下面是代码片段...抛出IllegalStateException。
java.lang.IllegalStateException:只允许一个连接接收订阅者。在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
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;
}
能够使用.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")
)