在Spring WebFlux中进行异步SOAP调用

Question

我有一个使用WebFlux和REST API的反应性Spring应用程序。每当用户调用我的API，我都需要调用一个公开WSDL的SOAP服务，执行一些操作并返回结果。

如何将对SOAP服务的调用与反应性WebFlux框架结合起来？

在我看来，我可以用两种不同的方式来做：

1. 使用WebFlux的WebClient构造并发送SOAP消息。
2. 在Mono/flux中使用WebServiceGatewaySupport包装同步调用。

public class MySoapClient extends WebServiceGatewaySupport {

    public QueryResponse execute() {
        Query query = new ObjectFactory().createQuery();
        // Further create and set the domain object here from the wsdl2java generated classes       
        return (QueryResponse) getWebServiceTemplate().marshalSendAndReceive(query);
    }
}

有人能分享从WebFlux控制器到进行SOAP调用并异步返回的完整示例吗？我觉得我错过了一些重要的东西。

Answer 1

我有同样的目标，但没有WSDL文件。作为输入，我有endpoint和XSD文件，它定义了我应该发送的请求方案。这是我的一段代码。

首先，让我们定义我们的SOPA WebClient bean（以避免每次需要调用时都创建它）

@Bean(name = "soapWebClient")
public WebClient soapWebClient(WebClient.Builder webClientBuilder) {
        String endpoint = environment.getRequiredProperty(ENDPOINT);
        log.info("Initializing SOAP Web Client ({}) bean...", endpoint);

        return webClientBuilder.baseUrl(endpoint)
                               .defaultHeader(CONTENT_TYPE, "application/soap+xml")
                               //if you have any time limitation put them here
                               .clientConnector(getWebClientConnector(SOAP_WEBCLIENT_CONNECT_TIMEOUT_SECONDS, SOAP_WEBCLIENT_IO_TIMEOUT_SECONDS))
                               //if you have any request/response size limitation put them here as well
                               .exchangeStrategies(ExchangeStrategies.builder()
                                                                     .codecs(configurer -> configurer.defaultCodecs()
                                                                                                     .maxInMemorySize(MAX_DATA_BUFFER_SIZE))
                                                                     .build())
                               .build();
}

public static ReactorClientHttpConnector getWebClientConnector(int connectTimeoutSeconds, int ioTimeoutSeconds) {
        TcpClient tcpClient = TcpClient.create()
                                       .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutSeconds * 1000)
                                       .doOnConnected(conn -> conn.addHandlerLast(new ReadTimeoutHandler(ioTimeoutSeconds))
                                                                  .addHandlerLast(new WriteTimeoutHandler(ioTimeoutSeconds)));
        return new ReactorClientHttpConnector(HttpClient.from(tcpClient));
}

现在您可以使用客户机进行SOAP调用，如下所示：

@Slf4j
@Component
public class SOAPClient {

    private final WebClient soapWebClient;

    public SOAPClient(@Qualifier("soapWebClient") WebClient soapWebClient) {
        this.soapWebClient = soapWebClient;
    }

    public Mono<Tuple2<HttpStatus, String>> send(String soapXML) {
        return Mono.just("Request:\n" + soapXML)
                   .doOnNext(log::info)
                   .flatMap(xml -> soapWebClient.post()
                                                .bodyValue(soapXML)
                                                .exchange()
                                                .doOnNext(res -> log.info("response status code: [{}]", res.statusCode()))
                                                .flatMap(res -> res.bodyToMono(String.class)
                                                                   .doOnNext(body -> log.info("Response body:\n{}", body))
                                                                   .map(b -> Tuples.of(res.statusCode(), b))
                                                                   .defaultIfEmpty(Tuples.of(res.statusCode(), "There is no data in the response"))))
                   .onErrorResume(ConnectException.class, e -> Mono.just(Tuples.of(SERVICE_UNAVAILABLE, "Failed to connect to server"))
                                                                   .doOnEach(logNext(t2 -> log.warn(t2.toString()))))
                   .onErrorResume(TimeoutException.class, e -> Mono.just(Tuples.of(GATEWAY_TIMEOUT, "There is no response from the server"))
                                                                   .doOnEach(logNext(t2 -> log.warn(t2.toString()))));
    }

}

这里要提到的一件重要事情是，soapxml显然应该是SOAP协议定义的格式。更具体地说，消息至少应该以soap:envelope标记开始和结束，并包含所有其他数据。另外，请注意您将要使用的SOAP协议版本，因为它定义了允许在信封内使用哪些标记，哪些不允许使用。我的代码是1.1，下面是它的规范https://www.w3.org/tr/2000/note-soap-20000508/#_toc478383494

干杯