0

我正在使用下面指定的 mule 流来发出 Web 服务的简单请求。请求成功发出，使用wireshark我看到了正确的响应。

在我包含一个输出文件端点（为了捕获对文件的响应）之前，我没有看到任何异常。即使有这个例外（包括在底部），当我查看生成的输出文件时，我也会看到我的数据，尽管在许多情况下它只是数据的一部分。

是什么导致了这个异常？

这是流程：

<flow name="TestAuditorClient_CheckerFlow1" doc:name="TestAuditorClient_CheckerFlow1">
    <quartz:inbound-endpoint jobName="gen" repeatInterval="10000" responseTimeout="10000" 
        doc:name="Quartz"> <quartz:event-generator-job /> </quartz:inbound-endpoint>
    <component class="wsdl.Payload" doc:name="Java"/>
    <cxf:jaxws-client operation="listTest"
        clientClass="server.TestService_Service" port="TestServicePort"
        wsdlLocation="http://192.168.66.7:8080/TestAuditorWebApp/TestService?wsdl"
        enableMuleSoapHeaders="true" doc:name="SOAPY" />
    <outbound-endpoint
        address="http://192.168.66.7:8080/TestAuditorWebApp/TestService"
        doc:name="HTTP"  mimeType="text/xml" />
    <logger message="Received HTTP Response #[payload]" level="INFO"
        doc:name="Logger" />
    <file:outbound-endpoint path="C:\tmp" outputPattern="#[function:datestamp:dd-MM-yy]_#[function:systime].txt" 
        responseTimeout="10000" doc:name="Output File" /> 
</flow>

这是一个例外：

INFO  2013-02-21 15:38:09,753 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02]    org.mule.api.processor.LoggerMessageProcessor: soap payload:      org.mule.module.cxf.transport.MuleUniversalConduit$1@69d7ef58
INFO  2013-02-21 15:38:09,769 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.transport.service.DefaultTransportServiceDescriptor: Loading default outbound transformer: org.mule.transport.http.transformers.ObjectToHttpClientMethodRequest
INFO  2013-02-21 15:38:09,769 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.transport.service.DefaultTransportServiceDescriptor: Loading default response transformer: org.mule.transport.http.transformers.MuleMessageToHttpResponse
INFO  2013-02-21 15:38:09,769 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.transport.service.DefaultTransportServiceDescriptor: Loading default outbound transformer: org.mule.transport.http.transformers.ObjectToHttpClientMethodRequest
INFO  2013-02-21 15:38:09,769 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.lifecycle.AbstractLifecycleManager: Initialising: 'HTTP_HTTPS.dispatcher.479670620'. Object is: HttpClientMessageDispatcher
INFO  2013-02-21 15:38:09,784 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.lifecycle.AbstractLifecycleManager: Starting: 'HTTP_HTTPS.dispatcher.479670620'. Object is: HttpClientMessageDispatcher
INFO  2013-02-21 15:38:09,878 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.api.processor.LoggerMessageProcessor: Received HTTP Response org.mule.transport.http.ReleasingInputStream@5e8aa5e1
INFO  2013-02-21 15:38:09,878 [[test_auditor_client4].connector.file.mule.default.dispatcher.01] org.mule.lifecycle.AbstractLifecycleManager: Initialising: 'connector.file.mule.default.dispatcher.784931034'. Object is: FileMessageDispatcher
INFO  2013-02-21 15:38:09,878 [[test_auditor_client4].connector.file.mule.default.dispatcher.01] org.mule.lifecycle.AbstractLifecycleManager: Starting: 'connector.file.mule.default.dispatcher.784931034'. Object is: FileMessageDispatcher
INFO  2013-02-21 15:38:09,894 [[test_auditor_client4].connector.file.mule.default.dispatcher.01] org.mule.transport.file.FileConnector: Writing file to: C:\tmp\21-02-13_1361479089894.txt
WARN  2013-02-21 15:38:09,894 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.apache.cxf.phase.PhaseInterceptorChain: Interceptor for {http://server/}TestService#{http://server/}listTest has thrown exception, unwinding now
java.lang.RuntimeException: Couldn't parse stream.
    at org.apache.cxf.staxutils.StaxUtils.createXMLStreamReader(StaxUtils.java:1181)
    at org.apache.cxf.interceptor.StaxInInterceptor.handleMessage(StaxInInterceptor.java:104)
    at org.apache.cxf.phase.PhaseInterceptorChain.doIntercept(PhaseInterceptorChain.java:263)
    at org.apache.cxf.endpoint.ClientImpl.onMessage(ClientImpl.java:799)
    at org.mule.module.cxf.transport.MuleUniversalConduit.dispatchMuleMessage(MuleUniversalConduit.java:280)
    at org.mule.module.cxf.transport.MuleUniversalConduit$2.handleMessage(MuleUniversalConduit.java:194)
    at org.apache.cxf.phase.PhaseInterceptorChain.doIntercept(PhaseInterceptorChain.java:263)
    at org.apache.cxf.endpoint.ClientImpl.doInvoke(ClientImpl.java:533)
    at org.apache.cxf.endpoint.ClientImpl.invoke(ClientImpl.java:463)
    at org.apache.cxf.endpoint.ClientImpl.invoke(ClientImpl.java:366)
    at org.apache.cxf.endpoint.ClientImpl.invoke(ClientImpl.java:319)
    at org.apache.cxf.frontend.ClientProxy.invokeSync(ClientProxy.java:88)
    at org.apache.cxf.jaxws.JaxWsClientProxy.invoke(JaxWsClientProxy.java:134)
    at $Proxy33.listTest(Unknown Source)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.mule.module.cxf.CxfOutboundMessageProcessor.doSendWithProxy(CxfOutboundMessageProcessor.java:187)
    at org.mule.module.cxf.CxfOutboundMessageProcessor.process(CxfOutboundMessageProcessor.java:129)
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:27)
    at org.mule.execution.MessageProcessorNotificationExecutionInterceptor.execute(MessageProcessorNotificationExecutionInterceptor.java:43)
    at org.mule.execution.MessageProcessorExecutionTemplate.execute(MessageProcessorExecutionTemplate.java:43)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:93)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:66)
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:27)
    at org.mule.execution.MessageProcessorExecutionTemplate.execute(MessageProcessorExecutionTemplate.java:43)
    at org.mule.processor.AbstractInterceptingMessageProcessorBase.processNext(AbstractInterceptingMessageProcessorBase.java:105)
    at org.mule.interceptor.AbstractEnvelopeInterceptor.process(AbstractEnvelopeInterceptor.java:55)
    at org.mule.processor.AsyncInterceptingMessageProcessor.processNextTimed(AsyncInterceptingMessageProcessor.java:118)
    at org.mule.processor.AsyncInterceptingMessageProcessor$AsyncMessageProcessorWorker$1.process(AsyncInterceptingMessageProcessor.java:186)
    at org.mule.processor.AsyncInterceptingMessageProcessor$AsyncMessageProcessorWorker$1.process(AsyncInterceptingMessageProcessor.java:179)
    at org.mule.execution.ExecuteCallbackInterceptor.execute(ExecuteCallbackInterceptor.java:20)
    at org.mule.execution.HandleExceptionInterceptor.execute(HandleExceptionInterceptor.java:34)
    at org.mule.execution.HandleExceptionInterceptor.execute(HandleExceptionInterceptor.java:18)
    at org.mule.execution.BeginAndResolveTransactionInterceptor.execute(BeginAndResolveTransactionInterceptor.java:58)
    at org.mule.execution.ResolvePreviousTransactionInterceptor.execute(ResolvePreviousTransactionInterceptor.java:48)
    at org.mule.execution.SuspendXaTransactionInterceptor.execute(SuspendXaTransactionInterceptor.java:54)
    at org.mule.execution.ValidateTransactionalStateInterceptor.execute(ValidateTransactionalStateInterceptor.java:44)
    at org.mule.execution.IsolateCurrentTransactionInterceptor.execute(IsolateCurrentTransactionInterceptor.java:44)
    at org.mule.execution.ExternalTransactionInterceptor.execute(ExternalTransactionInterceptor.java:52)
    at org.mule.execution.RethrowExceptionInterceptor.execute(RethrowExceptionInterceptor.java:32)
    at org.mule.execution.RethrowExceptionInterceptor.execute(RethrowExceptionInterceptor.java:17)
    at org.mule.execution.TransactionalErrorHandlingExecutionTemplate.execute(TransactionalErrorHandlingExecutionTemplate.java:113)
    at org.mule.execution.TransactionalErrorHandlingExecutionTemplate.execute(TransactionalErrorHandlingExecutionTemplate.java:34)
    at org.mule.processor.AsyncInterceptingMessageProcessor$AsyncMessageProcessorWorker.doRun(AsyncInterceptingMessageProcessor.java:178)
    at org.mule.work.AbstractMuleEventWork.run(AbstractMuleEventWork.java:43)
    at org.mule.work.WorkerContext.run(WorkerContext.java:311)
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: com.ctc.wstx.exc.WstxIOException: Attempted read on closed stream.
    at com.ctc.wstx.stax.WstxInputFactory.doCreateSR(WstxInputFactory.java:536)
    at com.ctc.wstx.stax.WstxInputFactory.createSR(WstxInputFactory.java:585)
    at com.ctc.wstx.stax.WstxInputFactory.createSR(WstxInputFactory.java:610)
    at com.ctc.wstx.stax.WstxInputFactory.createXMLStreamReader(WstxInputFactory.java:316)
    at org.apache.cxf.staxutils.StaxUtils.createXMLStreamReader(StaxUtils.java:1179)
    ... 50 more
Caused by: java.io.IOException: Attempted read on closed stream.
    at org.apache.commons.httpclient.AutoCloseInputStream.isReadAllowed(AutoCloseInputStream.java:183)
    at org.apache.commons.httpclient.AutoCloseInputStream.read(AutoCloseInputStream.java:107)
    at org.mule.model.streaming.DelegatingInputStream.read(DelegatingInputStream.java:63)
    at java.io.FilterInputStream.read(Unknown Source)
    at java.io.PushbackInputStream.read(Unknown Source)
    at com.ctc.wstx.io.BaseReader.readBytes(BaseReader.java:155)
    at com.ctc.wstx.io.UTF8Reader.loadMore(UTF8Reader.java:368)
    at com.ctc.wstx.io.UTF8Reader.read(UTF8Reader.java:111)
    at com.ctc.wstx.io.ReaderBootstrapper.initialLoad(ReaderBootstrapper.java:250)
    at com.ctc.wstx.io.ReaderBootstrapper.bootstrapInput(ReaderBootstrapper.java:133)
    at com.ctc.wstx.stax.WstxInputFactory.doCreateSR(WstxInputFactory.java:531)
    ... 54 more
ERROR 2013-02-21 15:38:09,972 [[test_auditor_client4].TestAuditorClient_CheckerFlow1.stage1.02] org.mule.exception.DefaultMessagingExceptionStrategy: 
********************************************************************************
Message               : Attempted read on closed stream.. Failed to route event via endpoint: org.mule.module.cxf.CxfOutboundMessageProcessor. Message payload is of type: PostMethod
Code                  : MULE_ERROR--2
--------------------------------------------------------------------------------
Exception stack is:
1. Attempted read on closed stream. (java.io.IOException)
  org.apache.commons.httpclient.AutoCloseInputStream:183 (null)
2. Attempted read on closed stream. (com.ctc.wstx.exc.WstxIOException)
  com.ctc.wstx.stax.WstxInputFactory:536 (null)
3. Couldn't parse stream. (java.lang.RuntimeException)
  org.apache.cxf.staxutils.StaxUtils:1181 (null)
4. Couldn't parse stream. (javax.xml.ws.soap.SOAPFaultException)
  org.apache.cxf.jaxws.JaxWsClientProxy:156 (http://java.sun.com/j2ee/sdk_1.3/techdocs/api/javax/xml/ws/soap/SOAPFaultException.html)
5. Attempted read on closed stream.. Failed to route event via endpoint: org.mule.module.cxf.CxfOutboundMessageProcessor. Message payload is of type: PostMethod (org.mule.api.transport.DispatchException)
  org.mule.module.cxf.CxfOutboundMessageProcessor:148 (http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/api/transport/DispatchException.html)
--------------------------------------------------------------------------------
Root Exception stack trace:
java.io.IOException: Attempted read on closed stream.
    at org.apache.commons.httpclient.AutoCloseInputStream.isReadAllowed(AutoCloseInputStream.java:183)
    at org.apache.commons.httpclient.AutoCloseInputStream.read(AutoCloseInputStream.java:107)
    at org.mule.model.streaming.DelegatingInputStream.read(DelegatingInputStream.java:63)
    + 3 more (set debug level logging or '-Dmule.verbose.exceptions=true' for everything)
********************************************************************************
4

2 回答 2

0

发生的情况是，HTTP 出站交互的结果首先由出站端点之后的内容处理，然后再传递回 CXF 客户端。此时，流式有效载荷已被消耗，CXF 无法再处理它了。

您可以尝试以下方法：

<flow name="TestAuditorClient_CheckerFlow1" doc:name="TestAuditorClient_CheckerFlow1">
    <quartz:inbound-endpoint jobName="gen" repeatInterval="10000" responseTimeout="10000" 
        doc:name="Quartz"> <quartz:event-generator-job /> </quartz:inbound-endpoint>
    <component class="wsdl.Payload" doc:name="Java"/>
    <http:outbound-endpoint
        address="http://192.168.66.7:8080/TestAuditorWebApp/TestService"
        doc:name="HTTP"  mimeType="text/xml">
        <cxf:jaxws-client operation="listTest"
            clientClass="server.TestService_Service" port="TestServicePort"
            wsdlLocation="http://192.168.66.7:8080/TestAuditorWebApp/TestService?wsdl"
            enableMuleSoapHeaders="true" doc:name="SOAPY" />
    </http:outbound-endpoint>
    <logger message="Received HTTP Response #[payload]" level="INFO"
        doc:name="Logger" />
    <file:outbound-endpoint path="C:\tmp"
        outputPattern="#[function:datestamp:dd-MM-yy]_#[function:systime].txt" 
        responseTimeout="10000" doc:name="Output File" /> 
</flow>

这个想法是将 CXF 交互本地化到 HTTP 端点。

请注意，我已将通用出站端点更改为 HTTP 端点，这不是强制性的，但至少您将获得对可用属性的更好支持。

另请注意，CXF 交互后流中的消息负载将是响应对象，而不是 SOAP 消息。您必须决定序列化策略（如 JAXB）以将其保存在文件中。

于 2013-02-21T21:55:03.127 回答
0

既然大卫已经花时间引导我朝着正确的方向前进，我将发布答案以供将来尝试此操作的人受益：

工作流程是：

<flow name="TestAuditorClient_CheckerFlow1" doc:name="TestAuditorClient_CheckerFlow1">
    <quartz:inbound-endpoint jobName="gen"
        repeatInterval="10000" responseTimeout="10000" doc:name="Quartz">
        <quartz:event-generator-job />
    </quartz:inbound-endpoint>
    <component class="wsdl.Payload" doc:name="Java" />
    <http:outbound-endpoint
        address="http://192.168.66.7:8080/TestAuditorWebApp/TestService"
        doc:name="HTTP" mimeType="text/xml">
        <cxf:jaxws-client operation="listTest"
            clientClass="server.TestService_Service" port="TestServicePort"
            wsdlLocation="http://192.168.66.7:8080/TestAuditorWebApp/TestService?wsdl"
            enableMuleSoapHeaders="true" doc:name="SOAPY" />
    </http:outbound-endpoint>
    <logger message="Received HTTP Response #[payload]" level="INFO"
        doc:name="Logger" />
    <component class="wsdl.Payload" doc:name="Java"/>
    <file:outbound-endpoint path="C:\tmp"
        outputPattern="#[function:datestamp:dd-MM-yy]_#[function:systime].txt"
        responseTimeout="10000" doc:name="Output File" />
</flow>

请注意，我在 HTTP 请求之前和请求之后都使用了 wsdl.Payload 类。这个单一的类用于生成请求和处理响应。

这是那个类：

package wsdl;

import server.ListTest;
import server.ListTestResponse;

public class Payload {

    public ListTest request() {
    System.out.println("request response");
    return new ListTest();
}

public String response(ListTestResponse listTestResponse) {
    System.out.println("process response");
    StringBuilder sb = new StringBuilder();

    List<Test> testList = listTestResponse.getReturn();
    Iterator<Test> testIt = testList.iterator();
    while(testIt.hasNext()) {
        Test test = testIt.next();
        sb.append(test.getName()).append(" ").append(test.getDescription()).append(", ");
    }
    return sb.toString();
}
}

瞧，现在骡子流程完全按照我的预期工作。

于 2013-02-22T15:10:47.687 回答