CXF：如何从CXF客户端重新发送原始请求？

Question

我是SOA开发环境中CXF的用户。

我想知道我的问题是否有CXF的解决方案。这是我的需要。我们开发了一个服务于JAXWSendpoint的webapp，endpoint实现包括通过拦截器分析请求，将请求中的数据存储到Java服务层的数据库中，并通过CXF客户端将原始请求重新发送到另一台服务器。关键是我们的一些请求包含DSIG签名(https://www.w3.org/TR/xmldsig-core/)或签名的SAML断言。我们需要在不改变CXFClient中的请求（例如代理）的情况下重新发送请求。CXF用于将封送处理的对象发送到服务器，但这样就不会发送原始流

有没有一种方法可以在不改变Java CXFClient的情况下从服务层重新发送传入请求（签名取决于请求的格式：空格、名称空间前缀、回车等）？我们更喜欢CXFClient，因为我们希望重用自制的CXF拦截器，该拦截器记录传出请求。

我们测试了一个拦截器，打算在将输出流发送到服务器之前用原始请求替换输出流，我们使用了以下答案：如何修改出站CXF请求的原始XML消息？，但我们仍然是KO，CXF仍然发送从编组对象生成的流。请参见下面的代码。

上下文：-CXF 2.7.18（JDK 6）和3.1.10（JDK 8）-平台：windows 7 64位/rhel 7 64位-Apache Tomcat 7-用于分析传入流量的Tcpdump

我们客户的代码示例：

    final Client cxfClient = org.apache.cxf.frontend.ClientProxy.getClient( portType );
    cxfClient.getInInterceptors().clear();
cxfClient.getOutInterceptors().clear();
cxfClient.getOutFaultInterceptors().clear();
cxfClient.getRequestContext().put(CustomStreamerInterceptor.STREAM_TO_SEND,
PhaseInterceptorChain.getCurrentMessage().getContent( InputStream.class ) );
cxfClient.getOutInterceptors().add( new CustomStreamerInterceptor() );
org.apache.cxf.transport.http.HTTPConduit http = (org.apache.cxf.transport.http.HTTPConduit) cxfClient.getConduit();
...
port.doSomething(someRequest);

CustomStreamrinterceptor：

package test;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.commons.io.IOUtils;
import org.apache.cxf.binding.soap.interceptor.SoapOutInterceptor.SoapOutEndingInterceptor;
import org.apache.cxf.helpers.LoadingByteArrayOutputStream;
import org.apache.cxf.interceptor.AbstractOutDatabindingInterceptor;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.io.CacheAndWriteOutputStream;
import org.apache.cxf.message.Message;
import org.apache.cxf.phase.Phase;
public class CustomStreamerInterceptor extends AbstractOutDatabindingInterceptor {
       public static final String STREAM_TO_SEND = "STREAM_TO_SEND";
       public CustomStreamerInterceptor () {
             super( Phase.WRITE_ENDING );
             addAfter( SoapOutEndingInterceptor.class.getName() );
       }
       @Override
       public void handleMessage( Message message ) throws Fault {
             try {
                    InputStream toSend = (InputStream) message.get( STREAM_TO_SEND );
                    if ( toSend != null ) {
                           toSend.reset();
                           LoadingByteArrayOutputStream lBos = new LoadingByteArrayOutputStream();
                           IOUtils.copy( toSend, lBos );
                           CacheAndWriteOutputStream cawos = (CacheAndWriteOutputStream) message.getContent( OutputStream.class );
                           cawos.resetOut( lBos, false );//fail !
                    }
             }
             catch ( Exception e ) {
                    throw new Fault( e );
             }
       }
}

谢谢你的帮助，这将非常有用。

Answer 1

我认为最好创建一个“经典”HTTP客户端，因为CXF不是为这种情况设计的，更常见的是使用它来列表从java到XML的对象...对你来说幸运的是，我用拦截器来处理这个问题。您可以编写一个拦截器，将CXF准备发送到服务器的输出流对象中的流复制出来。您需要小心拦截器的阶段和顺序，因为如果您使用Logging拦截器，您可能希望记录传出流。此拦截器可以完成这项工作，请确保它在任何日志拦截器之后运行。CXF 2.7.18的代码：

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.commons.lang.CharEncoding;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.interceptor.StaxOutInterceptor;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
import org.apache.cxf.phase.AbstractPhaseInterceptor;
import org.apache.cxf.phase.Phase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PassePlatClientInterceptorOut extends AbstractPhaseInterceptor<Message> {
private static final Logger LOG = LoggerFactory.getLogger( PassePlatClientInterceptorOut.class );

private final Exchange exchangeToReadFrom;

public PassePlatClientInterceptorOut( final Exchange exchange ) {
    super( Phase.PRE_STREAM );
    addBefore( StaxOutInterceptor.class.getName() );
    this.exchangeToReadFrom = exchange;
}

@Override
public void handleMessage( Message message ) {
    InputStream is = (InputStream) exchangeToReadFrom.get( PassePlatServerInterceptorIn.PASSE_PLAT_INTERCEPTOR_STREAM_SERVEUR );
    if ( is != null ) {
        message.put( org.apache.cxf.message.Message.ENCODING, CharEncoding.UTF_8 );
        OutputStream os = message.getContent( OutputStream.class );
        try {
            IOUtils.copy( is, os );
            is.close();
        }
        catch ( IOException e ) {
            LOG.error( "Error ...", e );
            message.setContent( Exception.class, e );
            throw new Fault( new Exception( "Error ...", e ) );
        }
        boolean everythingOK = message.getInterceptorChain().doInterceptStartingAt( message,
                org.apache.cxf.interceptor.MessageSenderInterceptor.MessageSenderEndingInterceptor.class.getName() );
        if ( !everythingOK ) {
            LOG.error( "Error ?" );
            throw new Fault( new Exception( "Error ..." ) );
        }
    }
}

}

要创建拦截器：

cxfClient.getInInterceptors().add( new PassePlatClientInterceptorIn( exchange ) );