0

I need to use the timeout value passed in the message header as the value for the collection aggregator. 

<collection-aggregator timeout="#[new Integer(header:SESSION:TIMEOUT)]"    failOnTimeout="false" doc:name="Collection Aggregator"/>

I tried to convert it to a number but it wouldn't take it. Can someone post an example of how I would do it?

Mule Config

<flow name="workIn">
    <http:inbound-endpoint exchange-pattern="request-response" address="http://localhost:5455/SQSearcher" doc:name="HTTP">
        <cxf:jaxws-service serviceClass="com.ace.st.tf.web.ws.sq.ISQSearcher" doc:name="SQ Searcher"/>
        <message-properties-transformer scope="session">
            <add-message-property value="1" key="DOBSearch" />
        </message-properties-transformer>
    </http:inbound-endpoint>

    <logger message="SQSearcher Inbound Payload-&gt; #[payload]" level="INFO" doc:name="Logger"/>       

    <message-properties-transformer scope="session">
        <add-message-property value="#[new Integer(payload.getTimeout())]" key="TIMEOUT" />                 
    </message-properties-transformer>

    <custom-transformer class="com.ace.st.tf.web.ws.sq.transformer.CCDMessageTransformer" doc:name="Java"/>

    <request-reply storePrefix="workStore" doc:name="Put message for processing">
        <vm:outbound-endpoint path="dispatchIn">
            <message-properties-transformer scope="outbound">
                <delete-message-property key="MULE_REPLYTO" />
                <add-message-property value="#[message:id]" key="MULE_CORRELATION_ID" />                    
            </message-properties-transformer>
        </vm:outbound-endpoint>

        <vm:inbound-endpoint path="dispatchOut" />
    </request-reply>

    <custom-transformer class="com.ace.st.tf.web.ws.sq.transformer.CcdSearchResultTransformer" doc:name="Java"/>

</flow>

<flow name="workDispatcher">
    <vm:inbound-endpoint path="dispatchIn" />
    <flow-ref name="workWorker" />
</flow>

<flow name="workWorker">        

    <message-properties-transformer scope="outbound" doc:name="Remember correlation">
        <add-message-property value="#[header:OUTBOUND:MULE_CORRELATION_ID]" key="MULE_CORRELATION_ID" />
        <add-message-property value="2" key="MULE_CORRELATION_GROUP_SIZE" />
    </message-properties-transformer>

    <all>
        <vm:outbound-endpoint path="vm.flow.1"/>
        <vm:outbound-endpoint path="vm.flow.2"/>
    </all>

</flow>

<flow name="workAggregator">
    <vm:inbound-endpoint path="vm.aggregate" />
    <logger message="payload to aggregate-&gt; #[payload]" level="INFO" doc:name="Logger"/>     
    <custom-aggregator class="com.ace.st.tf.web.ws.sq.aggregator.CustomAggregator" storePrefix="header:SESSION:TIMEOUT" failOnTimeout="false"/>
    <logger message="Aggregated Payload-&gt; #[payload]" level="INFO" doc:name="Logger"/>
    <vm:outbound-endpoint path="dispatchOut" />
</flow>

<flow name="CCDClient" doc:name="CCDClient">
    <vm:inbound-endpoint path="vm.flow.1" doc:name="VM"/>
    <logger message="CCDClient correlation id inbound-&gt; #[header:OUTBOUND:MULE_CORRELATION_GROUP_SIZE]" level="INFO" doc:name="Logger"/>

    <choice doc:name="Choice">
        <when expression="header:SESSION:DOBSearch=1">
            <outbound-endpoint address="http://localhost:4723/CcdSearcher" doc:name="Generic">
                <cxf:jaxws-client clientClass="com.ace.st.tf.web.ws.sq.ccd.client.CcdSearcher_Service" port="CcdSearcherPort" wsdlLocation="classpath:CcdSearcher.wsdl" operation="searchByNameDob"/>
            </outbound-endpoint>
        </when>
        <otherwise>
            <expression-component doc:name="set error message">payload="NOT A DOB SEARCH"</expression-component>
        </otherwise>
    </choice>
    <vm:outbound-endpoint path="vm.aggregate" doc:name="VM">
        <message-properties-transformer scope="outbound">
            <add-message-property value="#[header:INBOUND:MULE_CORRELATION_ID]" key="MULE_CORRELATION_ID" />
            <add-message-property value="#[header:INBOUND:MULE_CORRELATION_GROUP_SIZE]" key="MULE_CORRELATION_GROUP_SIZE" />
        </message-properties-transformer>
    </vm:outbound-endpoint>
</flow>

<flow name="ATspCisClient" doc:name="ATspCisClient">
    <vm:inbound-endpoint path="vm.flow.2" doc:name="VM"/>
    <choice doc:name="Choice">
        <when expression="header:SESSION:DOBSearch=1">
            <outbound-endpoint address="http://localhost:4724/ATspCisSearcher" doc:name="Generic">
                <cxf:jaxws-client clientClass="com.ace.st.tf.web.ws.sq.atspcis.client.AtspcisSearcher_Service" port="AtspcisSearcherPort" wsdlLocation="classpath:ATspCisSearcher.wsdl" operation="atspSearchByNameDob"/>
            </outbound-endpoint>
        </when>
        <otherwise>
            <expression-component doc:name="set error message">payload="NOT A DOB SEARCH"</expression-component>
        </otherwise>
    </choice>
    <vm:outbound-endpoint path="vm.aggregate" doc:name="VM">
        <message-properties-transformer scope="outbound">
            <add-message-property value="#[header:INBOUND:MULE_CORRELATION_ID]" key="MULE_CORRELATION_ID" />
            <add-message-property value="#[header:INBOUND:MULE_CORRELATION_GROUP_SIZE]" key="MULE_CORRELATION_GROUP_SIZE" />
        </message-properties-transformer>
    </vm:outbound-endpoint>
</flow>

CustomAggregator

public class CustomAggregator extends SimpleCollectionAggregator 
{

public String getStorePrefix() {
    return super.getStorePrefix();
}

public void setStorePrefix(String storePrefix) {
    super.setStorePrefix(storePrefix);
}

@Override
protected EventCorrelatorCallback getCorrelatorCallback(MuleContext muleContext) {

    return new CollectionCorrelatorCallback(muleContext,false,storePrefix) {
        @Override
        public MuleEvent aggregateEvents(EventGroup events) throws AggregationException {
            try {
                for (Iterator<MuleEvent> iterator = events.iterator(); iterator.hasNext();) {
                    MuleEvent event = iterator.next();
                    Object result = muleContext.getExpressionManager().evaluate("header:SESSION:TIMEOUT",event.getMessage());
                    setTimeout((java.lang.Integer)result);
                }
            }
            catch (ObjectStoreException e) {
                throw new AggregationException(events,null,e);
            }
            //return new DefaultMuleEvent(new DefaultMuleMessage(buffer.toString(), muleContext), events.getMessageCollectionEvent());
            return super.aggregateEvents(events);
        }
    };
}
}
4

1 回答 1

0

从我在代码库中看到的，timeout似乎不支持表达式。

您必须通过扩展org.mule.routing.SimpleCollectionAggregator、覆盖getTimeout()和引用它来创建自己的自定义聚合器<custom-aggregator class="...">。

不幸的是，尽管有激烈的骡夫行动，我无法让以下工作：

public class CustomAggregator extends SimpleCollectionAggregator
{
    private String timeoutExpression;

    public String getTimeoutExpression()
    {
        return timeoutExpression;
    }

    public void setTimeoutExpression(final String timeoutExpression)
    {
        this.timeoutExpression = timeoutExpression;
    }

    @Override
    public void initialise() throws InitialisationException
    {
        super.initialise();

        eventCorrelator = new EventCorrelator(getCorrelatorCallback(muleContext), next, messageInfoMapping,
            muleContext, flowConstruct.getName(), persistentStores, storePrefix)
        {
            @Override
            public long getTimeout()
            {
                return (Long) muleContext.getExpressionLanguage().evaluate(getTimeoutExpression(),
                    RequestContext.getEvent());
            }
        };
    }
}

配置：

    <custom-aggregator class="com.helloworld.CustomAggregator" timeout="60000">
        <spring:property name="timeoutExpression" value="#[Integer.valueOf(sessionVars.TIMEOUT)]" />
    </custom-aggregator>

原因是负责获取过期相关组的线程看不到当前的 Mule 事件上下文，因此无法根据动态属性计算任何内容。

因此，虽然可以彻底改革ExpiringGroupMonitoringThread使用动态值的机制，但我想说这是不切实际的，现在应该被认为是“不可能的”。

于 2012-11-29T18:08:17.763 回答