spring-boot - 响应式 Spring WebClient - 进行 SOAP 调用

Question

我希望从 Spring React WebClient 进行 SOAP 调用。我找不到任何相关文档。想知道会采取什么方法。现在我在想

1. 在单独的线程池上使用 JAXB 构造 SOAP 消息
2. 通过网络客户端将其转换为字符串进行调用
3. 在返回单独的 tp 时使用 jaxb 转换回 java。

有哪些缺点和其他方法？

Answer 1

这是一个 Spring Reactor 的工作示例:https://github.com/gungor/spring-webclient-soap

您需要使用自定义编码器将生成的 JAXB 类封装在肥皂信封中，如下所示，然后将其添加到 WebClient 的交换策略中。

package webclient.soap.encoding;

import org.reactivestreams.Publisher;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CodecException;
import org.springframework.core.codec.Encoder;
import org.springframework.core.codec.EncodingException;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.PooledDataBuffer;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.WebServiceMessageFactory;
import org.springframework.ws.client.core.WebServiceTemplate;
import org.springframework.ws.support.DefaultStrategiesHelper;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.MarshalException;
import javax.xml.bind.Marshaller;
import javax.xml.bind.annotation.XmlRootElement;
import javax.xml.bind.annotation.XmlType;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

public class Jaxb2SoapEncoder implements Encoder<Object> {

    private final JaxbContextContainer jaxbContexts = new JaxbContextContainer();

    @Override
    public boolean canEncode(ResolvableType elementType, MimeType mimeType) {
        Class<?> outputClass = elementType.toClass();
        return (outputClass.isAnnotationPresent(XmlRootElement.class) ||
                    outputClass.isAnnotationPresent(XmlType.class));

    }

    @Override
    public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
        return Flux.from(inputStream)
                .take(1)
                .concatMap(value -> encode(value, bufferFactory, elementType, mimeType, hints))
                .doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release);
    }

    @Override
    public List<MimeType> getEncodableMimeTypes() {
        return Arrays.asList( MimeTypeUtils.TEXT_XML );
    }



    private Flux<DataBuffer> encode(Object value ,
                                    DataBufferFactory bufferFactory,
                                    ResolvableType type,
                                    MimeType mimeType,
                                    Map<String, Object> hints){

        return Mono.fromCallable(() -> {
            boolean release = true;
            DataBuffer buffer = bufferFactory.allocateBuffer(1024);
            try {
                OutputStream outputStream = buffer.asOutputStream();
                Class<?> clazz = ClassUtils.getUserClass(value);
                Marshaller marshaller = initMarshaller(clazz);

                // here should be optimized
                DefaultStrategiesHelper helper = new DefaultStrategiesHelper(WebServiceTemplate.class);
                WebServiceMessageFactory messageFactory = helper.getDefaultStrategy(WebServiceMessageFactory.class);
                WebServiceMessage message = messageFactory.createWebServiceMessage();

                marshaller.marshal(value, message.getPayloadResult());
                message.writeTo(outputStream);

                release = false;
                return buffer;
            }
            catch (MarshalException ex) {
                throw new EncodingException(
                        "Could not marshal " + value.getClass() + " to XML", ex);
            }
            catch (JAXBException ex) {
                throw new CodecException("Invalid JAXB configuration", ex);
            }
            finally {
                if (release) {
                    DataBufferUtils.release(buffer);
                }
            }
        }).flux();
    }


    private Marshaller initMarshaller(Class<?> clazz) throws JAXBException {
        Marshaller marshaller = this.jaxbContexts.createMarshaller(clazz);
        marshaller.setProperty(Marshaller.JAXB_ENCODING, StandardCharsets.UTF_8.name());
        return marshaller;
    }
}

WebClient 配置

@Bean
    public WebClient webClient(){
        TcpClient tcpClient = TcpClient.create();

        tcpClient
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .doOnConnected(connection -> {
                    connection.addHandlerLast(new ReadTimeoutHandler(5000, TimeUnit.MILLISECONDS));
                    connection.addHandlerLast(new WriteTimeoutHandler(5000, TimeUnit.MILLISECONDS));
                });

        ExchangeStrategies exchangeStrategies = ExchangeStrategies.builder().codecs( clientCodecConfigurer -> {
            clientCodecConfigurer.customCodecs().encoder(new Jaxb2SoapEncoder());
        }).build();

        WebClient webClient = WebClient.builder()
                .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient).wiretap(true)))
                .exchangeStrategies( exchangeStrategies )
                .build();

        return webClient;
    }

网络客户端

public void call(GetCountryRequest getCountryRequest) throws SOAPException, ParserConfigurationException, IOException {

        webClient.post()
                .uri( soapServiceUrl )
                .contentType(MediaType.TEXT_XML)
                .body( Mono.just(getCountryRequest) , GetCountryRequest.class  )
                .retrieve()
                .onStatus(
                        HttpStatus::isError,
                        clientResponse ->
                                clientResponse
                                        .bodyToMono(String.class)
                                        .flatMap(
                                                errorResponseBody ->
                                                        Mono.error(
                                                                new ResponseStatusException(
                                                                        clientResponse.statusCode(),
                                                                        errorResponseBody))))

                .bodyToMono(GetCountryResponse.class)
                .doOnSuccess( (GetCountryResponse response) -> {
                    //handle success
                })
                .doOnError(ResponseStatusException.class, error -> {
                    //handle error
                })
                .subscribe();

    }