禁止 Mule ESB 阻止某些请求
Ban Mule ESB from blocking some requests
有时（非常频繁）mule ESB 会阻止一些 SOAP 请求。我怎样才能阻止 Mule 这样做呢？不知道是什么原因阻塞了
这就是我在 mule 配置中配置 http-listener 的方式：
<http:listener-config name="HTTP_listener" host="${MachineName}" port="${InboundPort}" doc:name="HTTP listener Configuration" />
<flow name="etryFlow" >
<http:listener allowedMethods="POST" config-ref="HTTP_listener" path="/{name}" doc:name="HTTP"/>
<set-payload value="#[app.registry.pairFactory.createInputStreamWithStringKeyInstance(message.inboundProperties['http.uri.params']['name'],payload))]" />
<jms:outbound-endpoint connector-ref="jms-connector" ref="UnsortedOrders" />
<logger level="INFO" doc:name="Logger" message="****Entry****" />
</flow>
注意：禁止的请求被转换为 BlockingTransferInputStream 而不是 BufferInputStream。
谢谢。
** 编辑 **
这是 createInputStreamWithStringKeyInstance 实现：
public static Pair<String, byte[]> createByteArrayWithStringKeyInstance(String key, byte value[]) throws IOException {
return new Pair<String, byte[]>(key, value);
}
和Pair.java是一个简单的Serializable Class（与JavaFX.Pair相同）
注意：输入是 SOAP XML
--编辑--
这是 mule 配置文件：
<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:file="http://www.mulesoft.org/schema/mule/file"
xmlns:http="http://www.mulesoft.org/schema/mule/http"
xmlns="http://www.mulesoft.org/schema/mule/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:spring="http://www.springframework.org/schema/beans"
xmlns:quartz="http://www.mulesoft.org/schema/mule/quartz"
xmlns:vm="http://www.mulesoft.org/schema/mule/vm"
xmlns:jms="http://www.mulesoft.org/schema/mule/jms"
xmlns:scripting="http://www.mulesoft.org/schema/mule/scripting"
xmlns:bpm="http://www.mulesoft.org/schema/mule/bpm"
xmlns:client="http://www.mulesoft.org/schema/mule/client"
xmlns:tracking="http://www.mulesoft.org/schema/mule/ee/tracking"
xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:management="http://www.mulesoft.org/schema/mule/management"
xsi:schemaLocation="http://www.mulesoft.org/schema/mule/file http://www.mulesoft.org/schema/mule/file/current/mule-file.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/3.2/mule.xsd
http://www.mulesoft.org/schema/mule/quartz http://www.mulesoft.org/schema/mule/quartz/3.2/mule-quartz.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/3.2/mule-vm.xsd
http://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/3.2/mule-jms.xsd
http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/3.2/mule-scripting.xsd
http://www.mulesoft.org/schema/mule/bpm http://www.mulesoft.org/schema/mule/bpm/3.2/mule-bpm.xsd
http://www.mulesoft.org/schema/mule/client http://www.mulesoft.org/schema/mule/client/3.2/mule-client.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.mulesoft.org/schema/mule/ee/tracking http://www.mulesoft.org/schema/mule/ee/tracking/current/mule-tracking-ee.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.mulesoft.org/schema/mule/management http://www.mulesoft.org/schema/mule/management/current/mule-management.xsd ">
<jms:activemq-connector name="jms-connector" brokerURL="${BrokerURL}" disableTemporaryReplyToDestinations="true" specification="1.1"/>
<jms:endpoint connector-ref="jms-connector" exchange-pattern="one-way" name="UnsortedOrders" queue="UnsortedOrders"/>
<jms:endpoint connector-ref="jms-connector" exchange-pattern="one-way" name="DestinationEMC" queue="DestinationEMC" />
<jms:endpoint connector-ref="jms-connector" exchange-pattern="one-way" name="DestinationOriginal" queue="DestinationOriginal"/>
<management:jmx-default-config>
<management:credentials>
<spring:entry key="admin" value="admin"/>
</management:credentials>
</management:jmx-default-config>
<spring:beans>
<context:property-placeholder location="classpath:Application.properties"/>
<spring:bean name="NoFactsBean" class="java.util.ArrayList" scope="singleton"/>
<spring:bean id="objectStore" class="org.mule.util.store.QueuePersistenceObjectStore"/>
<spring:bean id ="requestBean" class="com.hamgam.hit.esb.bean.RequestBean" factory-method="getInstance" scope="singleton"/>
</spring:beans>
<message-properties-transformer name="contentTypeTextXML">
<add-message-property key="Content-Type" value="text/xml;charset=UTF-8;" />
</message-properties-transformer>
<http:listener-config name="HTTP_listener" host="${MachineName}" port="${InboundPort}" doc:name="HTTP listener Configuration" />
<flow name="etryFlow">
<http:listener allowedMethods="POST" parseRequest="false" config-ref="HTTP_listener" path="/{name}" doc:name="HTTP" />
<object-to-byte-array-transformer/>
<expression-component>
payload = app.registry.requestBean.createServiceRequest(message.inboundProperties['http.uri.params']['name'],payload);
</expression-component>
<jms:outbound-endpoint connector-ref="jms-connector" ref="UnsortedOrders" />
<logger level="INFO" doc:name="Logger" message="****#[payload] Enters Entry****" />
</flow>
<flow name="consumeRequestsFlow" >
<jms:inbound-endpoint ref="UnsortedOrders" connector-ref="jms-connector" />
<logger level="INFO" doc:name="Logger" message="**** #[payload] Enters consumeRequestsFlow ****" />
<choice doc:name="Choice">
<when expression="#[app.registry.requestBean.isEMCRequired(payload)]">
<jms:outbound-endpoint connector-ref="jms-connector" ref="DestinationEMC" />
</when>
<otherwise>
<jms:outbound-endpoint connector-ref="jms-connector" ref="DestinationOriginal" />
</otherwise>
</choice>
</flow>
<flow name="ConsumeEmcQ">
<jms:inbound-endpoint connector-ref="jms-connector" ref="DestinationEMC" />
<logger level="INFO" doc:name="Logger" message="**** #[payload] Enters ConsumeEmcQ ****" />
<set-payload value="#[app.registry.requestBean.sendToEMC(payload)]" />
<until-successful objectStore-ref="objectStore" maxRetries="${MaximumRetry}" secondsBetweenRetries="${RetryInterval}">
<http:outbound-endpoint address="${ECMURL}" exchange-pattern="one-way">
<transformer ref="contentTypeTextXML"/>
</http:outbound-endpoint>
</until-successful>
</flow>
<flow name="ConsumeOrigQ">
<jms:inbound-endpoint connector-ref="jms-connector" ref="DestinationOriginal" />
<logger level="INFO" doc:name="Logger" message="**** #[payload] Enters ConsumeOriginalQ ****" />
<set-payload value="#[app.registry.requestBean.sendToDestination(payload)]" />
</flow>
<flow name = "EMCResponseFlow" >
<http:inbound-endpoint host="${MachineName}" port="${EMCResponsePort}" exchange-pattern="one-way">
<set-property propertyName="SOAPAction" value="#[message.inboundProperties['SOAPAction']]"/>
<transformer ref="contentTypeTextXML"/>
</http:inbound-endpoint>
<logger level="INFO" doc:name="Logger" message="***EMCResponseEntry****" />
</flow>
</mule>
假设 createInputStreamWithStringKeyInstance（你没有描述，所以很难帮助你）returns InputStream，Mule 最终构建了一个 JMS StreamMessage （根据 this call）。
从 Mule 代码中，您可以看到它使用以下逻辑将流式有效负载排入 StreamMessage：
while ((len = value.read(buffer)) != -1)
{
streamMessage.writeBytes(buffer, 0, len);
}
这很好，只要 createInputStreamWithStringKeyInstance 返回的流不在幕后做任何花哨的事情。
另请注意，JMS StreamMessage 可能不是您想要的。你读过its Javadoc吗？它说：
A StreamMessage object is used to send a stream of primitive types in the Java programming language.
我很确定您更想使用 JMS BytesMessage。如果有效负载是 byte[]（根据上面链接的转换逻辑），Mule 会创建这样一条消息，因此您宁愿在 jms:outbound-endpoint.
之前有一个 <object-to-byte-array-transformer/>
有时（非常频繁）mule ESB 会阻止一些 SOAP 请求。我怎样才能阻止 Mule 这样做呢？不知道是什么原因阻塞了
这就是我在 mule 配置中配置 http-listener 的方式：
<http:listener-config name="HTTP_listener" host="${MachineName}" port="${InboundPort}" doc:name="HTTP listener Configuration" />
<flow name="etryFlow" >
<http:listener allowedMethods="POST" config-ref="HTTP_listener" path="/{name}" doc:name="HTTP"/>
<set-payload value="#[app.registry.pairFactory.createInputStreamWithStringKeyInstance(message.inboundProperties['http.uri.params']['name'],payload))]" />
<jms:outbound-endpoint connector-ref="jms-connector" ref="UnsortedOrders" />
<logger level="INFO" doc:name="Logger" message="****Entry****" />
</flow>
注意：禁止的请求被转换为 BlockingTransferInputStream 而不是 BufferInputStream。
谢谢。
** 编辑 **
这是 createInputStreamWithStringKeyInstance 实现：
public static Pair<String, byte[]> createByteArrayWithStringKeyInstance(String key, byte value[]) throws IOException {
return new Pair<String, byte[]>(key, value);
}
和Pair.java是一个简单的Serializable Class（与JavaFX.Pair相同） 注意：输入是 SOAP XML
--编辑-- 这是 mule 配置文件：
<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:file="http://www.mulesoft.org/schema/mule/file"
xmlns:http="http://www.mulesoft.org/schema/mule/http"
xmlns="http://www.mulesoft.org/schema/mule/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:spring="http://www.springframework.org/schema/beans"
xmlns:quartz="http://www.mulesoft.org/schema/mule/quartz"
xmlns:vm="http://www.mulesoft.org/schema/mule/vm"
xmlns:jms="http://www.mulesoft.org/schema/mule/jms"
xmlns:scripting="http://www.mulesoft.org/schema/mule/scripting"
xmlns:bpm="http://www.mulesoft.org/schema/mule/bpm"
xmlns:client="http://www.mulesoft.org/schema/mule/client"
xmlns:tracking="http://www.mulesoft.org/schema/mule/ee/tracking"
xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:management="http://www.mulesoft.org/schema/mule/management"
xsi:schemaLocation="http://www.mulesoft.org/schema/mule/file http://www.mulesoft.org/schema/mule/file/current/mule-file.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/3.2/mule.xsd
http://www.mulesoft.org/schema/mule/quartz http://www.mulesoft.org/schema/mule/quartz/3.2/mule-quartz.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/3.2/mule-vm.xsd
http://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/3.2/mule-jms.xsd
http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/3.2/mule-scripting.xsd
http://www.mulesoft.org/schema/mule/bpm http://www.mulesoft.org/schema/mule/bpm/3.2/mule-bpm.xsd
http://www.mulesoft.org/schema/mule/client http://www.mulesoft.org/schema/mule/client/3.2/mule-client.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.mulesoft.org/schema/mule/ee/tracking http://www.mulesoft.org/schema/mule/ee/tracking/current/mule-tracking-ee.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.mulesoft.org/schema/mule/management http://www.mulesoft.org/schema/mule/management/current/mule-management.xsd ">
<jms:activemq-connector name="jms-connector" brokerURL="${BrokerURL}" disableTemporaryReplyToDestinations="true" specification="1.1"/>
<jms:endpoint connector-ref="jms-connector" exchange-pattern="one-way" name="UnsortedOrders" queue="UnsortedOrders"/>
<jms:endpoint connector-ref="jms-connector" exchange-pattern="one-way" name="DestinationEMC" queue="DestinationEMC" />
<jms:endpoint connector-ref="jms-connector" exchange-pattern="one-way" name="DestinationOriginal" queue="DestinationOriginal"/>
<management:jmx-default-config>
<management:credentials>
<spring:entry key="admin" value="admin"/>
</management:credentials>
</management:jmx-default-config>
<spring:beans>
<context:property-placeholder location="classpath:Application.properties"/>
<spring:bean name="NoFactsBean" class="java.util.ArrayList" scope="singleton"/>
<spring:bean id="objectStore" class="org.mule.util.store.QueuePersistenceObjectStore"/>
<spring:bean id ="requestBean" class="com.hamgam.hit.esb.bean.RequestBean" factory-method="getInstance" scope="singleton"/>
</spring:beans>
<message-properties-transformer name="contentTypeTextXML">
<add-message-property key="Content-Type" value="text/xml;charset=UTF-8;" />
</message-properties-transformer>
<http:listener-config name="HTTP_listener" host="${MachineName}" port="${InboundPort}" doc:name="HTTP listener Configuration" />
<flow name="etryFlow">
<http:listener allowedMethods="POST" parseRequest="false" config-ref="HTTP_listener" path="/{name}" doc:name="HTTP" />
<object-to-byte-array-transformer/>
<expression-component>
payload = app.registry.requestBean.createServiceRequest(message.inboundProperties['http.uri.params']['name'],payload);
</expression-component>
<jms:outbound-endpoint connector-ref="jms-connector" ref="UnsortedOrders" />
<logger level="INFO" doc:name="Logger" message="****#[payload] Enters Entry****" />
</flow>
<flow name="consumeRequestsFlow" >
<jms:inbound-endpoint ref="UnsortedOrders" connector-ref="jms-connector" />
<logger level="INFO" doc:name="Logger" message="**** #[payload] Enters consumeRequestsFlow ****" />
<choice doc:name="Choice">
<when expression="#[app.registry.requestBean.isEMCRequired(payload)]">
<jms:outbound-endpoint connector-ref="jms-connector" ref="DestinationEMC" />
</when>
<otherwise>
<jms:outbound-endpoint connector-ref="jms-connector" ref="DestinationOriginal" />
</otherwise>
</choice>
</flow>
<flow name="ConsumeEmcQ">
<jms:inbound-endpoint connector-ref="jms-connector" ref="DestinationEMC" />
<logger level="INFO" doc:name="Logger" message="**** #[payload] Enters ConsumeEmcQ ****" />
<set-payload value="#[app.registry.requestBean.sendToEMC(payload)]" />
<until-successful objectStore-ref="objectStore" maxRetries="${MaximumRetry}" secondsBetweenRetries="${RetryInterval}">
<http:outbound-endpoint address="${ECMURL}" exchange-pattern="one-way">
<transformer ref="contentTypeTextXML"/>
</http:outbound-endpoint>
</until-successful>
</flow>
<flow name="ConsumeOrigQ">
<jms:inbound-endpoint connector-ref="jms-connector" ref="DestinationOriginal" />
<logger level="INFO" doc:name="Logger" message="**** #[payload] Enters ConsumeOriginalQ ****" />
<set-payload value="#[app.registry.requestBean.sendToDestination(payload)]" />
</flow>
<flow name = "EMCResponseFlow" >
<http:inbound-endpoint host="${MachineName}" port="${EMCResponsePort}" exchange-pattern="one-way">
<set-property propertyName="SOAPAction" value="#[message.inboundProperties['SOAPAction']]"/>
<transformer ref="contentTypeTextXML"/>
</http:inbound-endpoint>
<logger level="INFO" doc:name="Logger" message="***EMCResponseEntry****" />
</flow>
</mule>
假设 createInputStreamWithStringKeyInstance（你没有描述，所以很难帮助你）returns InputStream，Mule 最终构建了一个 JMS StreamMessage （根据 this call）。
从 Mule 代码中，您可以看到它使用以下逻辑将流式有效负载排入 StreamMessage：
while ((len = value.read(buffer)) != -1)
{
streamMessage.writeBytes(buffer, 0, len);
}
这很好，只要 createInputStreamWithStringKeyInstance 返回的流不在幕后做任何花哨的事情。
另请注意，JMS StreamMessage 可能不是您想要的。你读过its Javadoc吗？它说：
A StreamMessage object is used to send a stream of primitive types in the Java programming language.
我很确定您更想使用 JMS BytesMessage。如果有效负载是 byte[]（根据上面链接的转换逻辑），Mule 会创建这样一条消息，因此您宁愿在 jms:outbound-endpoint.
<object-to-byte-array-transformer/>