Aggregate two JMS messages from two different consumers

Question

I have two jms consumers, each in a different flow. I want to use another flow to aggregate the messages of thoe two messages. and also need to keep the correlation Ids as i need to split the payload and send back the messages.

    <flow name="integration-consumer-client1" doc:name="integration-consumer-client1">
    <jms:inbound-endpoint doc:name="JMS" connector-ref="Active_MQ" queue="client1.publish"/>
    <logger message="Consumes Client One = #[payload]" level="INFO" doc:name="Logger"/>
    <logger message="Client One Correlation = #[message.correlationId]" level="INFO" doc:name="Logger"/>
    <vm:outbound-endpoint exchange-pattern="one-way" path="client1" doc:name="VM"/>
</flow>
<flow name="integration-consumer-client2" doc:name="integration-consumer-client2">
    <jms:inbound-endpoint doc:name="JMS" connector-ref="Active_MQ" queue="client2.publish"/>
    <logger message="Consumes Client Two = #[payload]" level="INFO" doc:name="Logger"/>
    <logger message="Client Two Correlation = #[message.correlationId]" level="INFO" doc:name="Logger"/>
    <vm:outbound-endpoint exchange-pattern="one-way" path="client2" doc:name="VM"/>
</flow>
<flow name="integration-internetsolutionsFlow1" doc:name="integration-internetsolutionsFlow1">
    <scatter-gather doc:name="Scatter-Gather">
        <vm:inbound-endpoint exchange-pattern="one-way" path="client1" doc:name="VM"/>
        <vm:inbound-endpoint exchange-pattern="one-way" path="client2" doc:name="VM"/>
    </scatter-gather>
</flow>

I tried to use the scatter gather with two inbound VMs but get the following error:

Caused by: org.xml.sax.SAXParseException: cvc-complex-type.2.4.a: Invalid content was found starting with element 'vm:inbound-endpoint'. One of '{"http://www.mulesoft.org/schema/mule/core":annotations, "http://www.mulesoft.org/schema/mule/core":custom-aggregation-strategy, "http://www.mulesoft.org/schema/mule/core":threading-profile, "http://www.mulesoft.org/schema/mule/core":abstract-message-processor, "http://www.mulesoft.org/schema/mule/core":abstract-outbound-endpoint, "http://www.mulesoft.org/schema/mule/core":abstract-mixed-content-message-processor}' is expected. at org.apache.xerces.util.ErrorHandlerWrapper.createSAXParseException(Unknown Source)

Anirban Sen Chowdhary Anirban Sen Chowdhary · Accepted Answer · 2014-11-16T18:55:23

The reason of the error is integration-internetsolutionsFlow1 doesn't have any inbound endpoint ..

What you can do is following :-

Use vm:outbound-endpoint to same path from both the flow integration-consumer-client1 and integration-consumer-client2

Then in integration-internetsolutionsFlow1

     <flow name="integration-internetsolutionsFlow1" doc:name="integration-internetsolutionsFlow1">
      <vm:inbound-endpoint exchange-pattern="request-response" path="Your Path" doc:name="VM" connector-ref="vmConnector" />
    <logger level="INFO" message="#[message.payload]" doc:name="Logger"/>
<!-- you don't require a setter-getter here -->
    </flow>

The path of outbound VM from both the flow integration-consumer-client1 and integration-consumer-client2 should be same..

You no need a scatter-gather here .. since both the flow will disptch the payload to same VM path .. It will be receive by VM inbound endpoint

Aggregate two JMS messages from two different consumers

1 Answers