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

Question

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

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

有没有办法从 Java CXFClient 重新发送来自服务层的传入请求而不更改它（签名取决于请求的格式：空格、命名空间前缀、回车...）？我们更喜欢 CXFClient，因为我们想重用我们自制的记录传出请求的 CXF 拦截器。

我们测试了一个拦截器，打算在发送到服务器之前用原始请求替换 outputStream，我们使用了这个答案：How To Modify The Raw XML message of an Outbound CXF Request?，但我们仍然是 KO，CXF 仍然发送由编组对象生成的流.请参阅下面的代码。

上下文： - CXF 2.7.18 (JDK 6) 和 3.1.10 (JDK 8) - 平台：windows 7 64bit/rhel 7 64bit - 阿帕奇雄猫 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);

---

CustomStreamerInterceptor：

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