来自 jdbc 源的 Mule 消息丰富器答案

【问题标题】：Mule message enricher from jdbc source来自 jdbc 源的 Mule 消息丰富器
【发布时间】：2011-12-28 18:15:59
【问题描述】：

我想用两个变量 LAST_POLL_START 和 LAST_POLL_END 来丰富当前消息。然后这些变量将用于构建另一个 jdbc 查询。

但是，我不明白应该如何使用浓缩器。我尝试过的代码如下，它的灵感来自here 和Mule's blog 上的代码。在我看来，这应该有效。我正在使用 mule-3.2.0，jdbc 查询有效并返回正确的值。

<flow name="MasterFlow" processingStrategy="synchronous">
    <quartz:inbound-endpoint jobName="cronJobPoolTime" cronExpression="0 0/1 * * * ?">
        <quartz:endpoint-polling-job>
            <quartz:job-endpoint ref="jdbc_quartzDummy"/>
        </quartz:endpoint-polling-job>
    </quartz:inbound-endpoint>      
    <flow-ref name="GetPollingTimes" />
    <!-- Do other things -->
</flow>

<!-- Below is needed when using Quartz trigger -->
<jdbc:endpoint name="jdbc_quartzDummy" connector-ref="tabuConnector" queryKey="quartzDummy" />

<flow name="GetPollingTimes">
    <enricher>
        <jdbc:outbound-endpoint queryKey="getPollTimes" exchange-pattern="request-response" />
        <enrich target="#[variable:last_poll_start]" source="#[map-payload:LAST_POLL_START]"/>
        <enrich target="#[variable:last_poll_end]" source="#[map-payload:LAST_POLL_END]"/>
    </enricher>
    <logger level="INFO" message="last_poll_start = #[variable:last_poll_start]" />
</flow>

错误信息是：

WARN  2011-12-27 15:38:00,831 [scheduler-ESB_Worker-1] org.mule.DefaultMuleMessage: setProperty(key, value) called with null value; removing key: last_poll_start; please report the following stack trace to dev@mule.codehaus.org
java.lang.Throwable
    at org.mule.DefaultMuleMessage.setProperty(DefaultMuleMessage.java:456)
    at org.mule.expression.VariableExpressionEnricher.enrich(VariableExpressionEnricher.java:24)
    at org.mule.expression.DefaultExpressionManager.enrich(DefaultExpressionManager.java:248)
    at org.mule.expression.DefaultExpressionManager.enrich(DefaultExpressionManager.java:237)
    at org.mule.enricher.MessageEnricher.enrich(MessageEnricher.java:69)
    at org.mule.enricher.MessageEnricher.process(MessageEnricher.java:43)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.AbstractInterceptingMessageProcessorBase.processNext(AbstractInterceptingMessageProcessorBase.java:105)
    at org.mule.processor.AsyncInterceptingMessageProcessor.process(AsyncInterceptingMessageProcessor.java:90)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.AbstractInterceptingMessageProcessorBase.processNext(AbstractInterceptingMessageProcessorBase.java:105)
    at org.mule.interceptor.AbstractEnvelopeInterceptor.process(AbstractEnvelopeInterceptor.java:55)
    at org.mule.processor.AbstractInterceptingMessageProcessorBase.processNext(AbstractInterceptingMessageProcessorBase.java:105)
    at org.mule.processor.AbstractFilteringMessageProcessor.process(AbstractFilteringMessageProcessor.java:41)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.doProcess(InterceptingChainLifecycleWrapper.java:56)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.process(InterceptingChainLifecycleWrapper.java:87)
    at org.mule.construct.Flow.process(Flow.java:64)
    at org.mule.config.spring.factories.FlowRefFactoryBean$1.process(FlowRefFactoryBean.java:44)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.AbstractInterceptingMessageProcessorBase.processNext(AbstractInterceptingMessageProcessorBase.java:105)
    at org.mule.interceptor.AbstractEnvelopeInterceptor.process(AbstractEnvelopeInterceptor.java:55)
    at org.mule.processor.AbstractInterceptingMessageProcessorBase.processNext(AbstractInterceptingMessageProcessorBase.java:105)
    at org.mule.processor.AbstractFilteringMessageProcessor.process(AbstractFilteringMessageProcessor.java:41)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.doProcess(InterceptingChainLifecycleWrapper.java:56)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.process(InterceptingChainLifecycleWrapper.java:87)
    at org.mule.construct.AbstractPipeline$1.process(AbstractPipeline.java:138)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.doProcess(InterceptingChainLifecycleWrapper.java:56)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.process(InterceptingChainLifecycleWrapper.java:87)
    at org.mule.processor.chain.DefaultMessageProcessorChain.doProcess(DefaultMessageProcessorChain.java:99)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.doProcess(InterceptingChainLifecycleWrapper.java:56)
    at org.mule.processor.chain.AbstractMessageProcessorChain.process(AbstractMessageProcessorChain.java:64)
    at org.mule.processor.chain.InterceptingChainLifecycleWrapper.process(InterceptingChainLifecycleWrapper.java:87)
    at org.mule.transport.AbstractMessageReceiver.routeMessage(AbstractMessageReceiver.java:190)
    at org.mule.transport.AbstractMessageReceiver.routeMessage(AbstractMessageReceiver.java:163)
    at org.mule.transport.AbstractMessageReceiver.routeMessage(AbstractMessageReceiver.java:150)
    at org.mule.transport.quartz.jobs.EndpointPollingJob$1.doInTransaction(EndpointPollingJob.java:162)
    at org.mule.transport.quartz.jobs.EndpointPollingJob$1.doInTransaction(EndpointPollingJob.java:125)
    at org.mule.transaction.TransactionTemplate.execute(TransactionTemplate.java:45)
    at org.mule.transport.quartz.jobs.EndpointPollingJob.doExecute(EndpointPollingJob.java:169)
    at org.mule.transport.quartz.jobs.AbstractJob.execute(AbstractJob.java:36)
    at org.quartz.core.JobRunShell.run(JobRunShell.java:202)
    at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:534)

将丰富源更改为#[payload]：

<enrich target="#[variable:last_poll_start]" source="#[payload]"/>
<logger level="INFO" message="last_poll_start = #[variable:last_poll_start]" />

产生（来自日志文件）：

org.mule.api.processor.LoggerMessageProcessor: last_poll_start = [{LAST_POLL_START=2011-12-21, LAST_POLL_END=2000-01-01}]

【问题讨论】：

您确定地图有效负载中键的大小写吗？ map-payload:LAST_POLL_START 需要在地图有效负载中使用大写的 DB 字段名称。
我已更新问题以显示使用 source="#[payload]" 的输出。这就是相信它是大写的原因。
此外，source="#[payload:LAST_POLL_START]" 给出：org.mule.api.MuleRuntimeException：无法处理表达式评估“LAST_POLL_START”

标签： esb mule

【解决方案1】：

受上一个答案的启发，我注意到可以在丰富器中使用 groovy 语法。这是解决方案，如果其他人遇到这个问题。

<flow name="GetPollingTimes">
    <enricher>
        <jdbc:outbound-endpoint queryKey="getPollTimes" exchange-pattern="request-response" />
        <enrich target="#[variable:last_poll_start]" source="#[groovy:payload.last_poll_start]"/>
        <enrich target="#[variable:last_poll_end]" source="#[groovy:payload.last_poll_end]"/>
    </enricher>
</flow>

【讨论】：

所以有效载荷是一个带有小写键的映射。工作吗？
以下不起作用：#[payload:last_poll_start]、#[map-payload:last_poll_start]、#[map-payload:LAST_POLL_START]、#[payload:LAST_POLL_START]。对我来说，这很奇怪。

【解决方案2】：

我找到了这个问题的部分解决方案，它使用 groovy 脚本。然而，这感觉就像是某种我宁愿没有的 hack：

<flow name="GetPollingTimes">
    <jdbc:outbound-endpoint queryKey="getPollTimes" exchange-pattern="request-response" />
    <script:transformer>
        <script:script engine="groovy">
            <script:text>
                message.setInvocationProperty('last_poll_start', payload.last_poll_start)
                message.setInvocationProperty('last_poll_end', payload.last_poll_end)
                return payload
            </script:text>
        </script:script>
    </script:transformer>
</flow>

【讨论】：